From 1da11ed1c3ddefcb8e63aa37a3bcceb93d8f3223 Mon Sep 17 00:00:00 2001 From: Yuhong Cheng Date: Fri, 20 Sep 2019 15:05:03 -0700 Subject: [PATCH 1/6] Add interface to support specific configs --- .../org/apache/samza/config/SystemConfig.java | 2 + .../samza/util/CoordinatorStreamUtil.scala | 20 ++++---- .../util/CoordinatorStreamUtilFactory.java | 27 +++++++++++ .../samza/util/DefaultCoordinatorStream.java | 48 +++++++++++++++++++ .../util/TestCoordinatorStreamUtil.scala | 18 +++++++ 5 files changed, 103 insertions(+), 12 deletions(-) create mode 100644 samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtilFactory.java create mode 100644 samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStream.java diff --git a/samza-core/src/main/java/org/apache/samza/config/SystemConfig.java b/samza-core/src/main/java/org/apache/samza/config/SystemConfig.java index 93a0c32974..52559cb6db 100644 --- a/samza-core/src/main/java/org/apache/samza/config/SystemConfig.java +++ b/samza-core/src/main/java/org/apache/samza/config/SystemConfig.java @@ -37,6 +37,8 @@ public class SystemConfig extends MapConfig { private static final String SYSTEMS_PREFIX = "systems."; public static final String SYSTEM_ID_PREFIX = SYSTEMS_PREFIX + "%s."; + public static final String LI_SYSTEM_ID_PREFIX = SYSTEMS_PREFIX + "li."; + private static final String SYSTEM_FACTORY_SUFFIX = ".samza.factory"; public static final String SYSTEM_FACTORY_FORMAT = SYSTEMS_PREFIX + "%s" + SYSTEM_FACTORY_SUFFIX; diff --git a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala index e38359db9f..bc1d61e653 100644 --- a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala +++ b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala @@ -23,7 +23,6 @@ package org.apache.samza.util import java.util -import org.apache.commons.lang3.StringUtils import org.apache.samza.SamzaException import org.apache.samza.config._ import org.apache.samza.coordinator.metadatastore.{CoordinatorStreamStore, NamespaceAwareCoordinatorStreamStore} @@ -33,24 +32,21 @@ import org.apache.samza.system.{StreamSpec, SystemAdmin, SystemFactory, SystemSt import org.apache.samza.util.ScalaJavaUtil.JavaOptionals import scala.collection.JavaConverters._ -import scala.collection.immutable.Map object CoordinatorStreamUtil extends Logging { + val COORDINATOR_STREAM_FACTORY = "coordinatorstreamFactory" + val DEFAULT_COORDINATOR_STREAM = "org.apache.samza.util.DefaultCoordinatorStream" + /** * Given a job's full config object, build a subset config which includes * only the job name, job id, and system config for the coordinator stream. */ def buildCoordinatorStreamConfig(config: Config) = { - val jobConfig = new JobConfig(config) - val (jobName, jobId) = getJobNameAndId(jobConfig) - // Build a map with just the system config and job.name/job.id. This is what's required to start the JobCoordinator. - val map = config.subset(SystemConfig.SYSTEM_ID_PREFIX format jobConfig.getCoordinatorSystemName, false).asScala ++ - Map[String, String]( - JobConfig.JOB_NAME -> jobName, - JobConfig.JOB_ID -> jobId, - JobConfig.JOB_COORDINATOR_SYSTEM -> jobConfig.getCoordinatorSystemName, - JobConfig.MONITOR_PARTITION_CHANGE_FREQUENCY_MS -> String.valueOf(jobConfig.getMonitorPartitionChangeFrequency)) - new MapConfig(map.asJava) + val buildConfigFactory = config.get(COORDINATOR_STREAM_FACTORY, DEFAULT_COORDINATOR_STREAM) + val coordinatorSystemConfig = Class.forName(buildConfigFactory).newInstance().asInstanceOf[CoordinatorStreamUtilFactory].buildCoordinatorStreamConfig(config) + + new MapConfig((coordinatorSystemConfig)); + } /** diff --git a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtilFactory.java b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtilFactory.java new file mode 100644 index 0000000000..376b267020 --- /dev/null +++ b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtilFactory.java @@ -0,0 +1,27 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.samza.util; + +import org.apache.samza.config.Config; + + +public interface CoordinatorStreamUtilFactory { + Config buildCoordinatorStreamConfig(Config config); +} diff --git a/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStream.java b/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStream.java new file mode 100644 index 0000000000..6073831074 --- /dev/null +++ b/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStream.java @@ -0,0 +1,48 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.samza.util; + +import java.util.HashMap; +import java.util.Map; +import org.apache.samza.config.Config; +import org.apache.samza.config.JobConfig; +import org.apache.samza.config.MapConfig; +import org.apache.samza.config.SystemConfig; + + +public class DefaultCoordinatorStream implements CoordinatorStreamUtilFactory { + @Override + public Config buildCoordinatorStreamConfig(Config config) { + Config jobConfig = new JobConfig(config); + String jobName = ((JobConfig) jobConfig).getName().toString(); + String jobId = ((JobConfig) jobConfig).getJobId(); + + // Build a map with just the system config and job.name/job.id. This is what's required to start the JobCoordinator. + Map map = config.subset(String.format(SystemConfig.SYSTEM_ID_PREFIX, ((JobConfig) jobConfig).getCoordinatorSystemName()), false); + Map addConfig = new HashMap<>(); + addConfig.put(JobConfig.JOB_NAME, jobName); + addConfig.put(((JobConfig) jobConfig).JOB_ID, jobId); + addConfig.put(JobConfig.JOB_COORDINATOR_SYSTEM, ((JobConfig) jobConfig).getCoordinatorSystemName()); + addConfig.put(((JobConfig) jobConfig).MONITOR_PARTITION_CHANGE_FREQUENCY_MS, String.valueOf(((JobConfig) jobConfig).getMonitorPartitionChangeFrequency())); + + addConfig.putAll(map); + return new MapConfig(addConfig); + } +} diff --git a/samza-core/src/test/scala/org/apache/samza/util/TestCoordinatorStreamUtil.scala b/samza-core/src/test/scala/org/apache/samza/util/TestCoordinatorStreamUtil.scala index eb035960d9..edf9ecb730 100644 --- a/samza-core/src/test/scala/org/apache/samza/util/TestCoordinatorStreamUtil.scala +++ b/samza-core/src/test/scala/org/apache/samza/util/TestCoordinatorStreamUtil.scala @@ -27,6 +27,7 @@ import org.apache.samza.system.{StreamSpec, SystemAdmin, SystemStream} import org.junit.{Assert, Test} import org.mockito.Matchers.any import org.mockito.Mockito +import org.apache.samza.config.MapConfig class TestCoordinatorStreamUtil { @@ -40,6 +41,23 @@ class TestCoordinatorStreamUtil { Mockito.verify(systemAdmin).createStream(any(classOf[StreamSpec])) } + @Test + def testBuildCoordinatorStreamConfig: Unit = { + val addConfig = new util.HashMap[String, String] + addConfig.put("job.name", "test-job-name") + addConfig.put("job.id", "i001") + addConfig.put("job.coordinator.system", "samzatest") + addConfig.put("systems.samzatest.test","test") + addConfig.put("test.only","nothing") + val config = new MapConfig(addConfig) + + var configMap = CoordinatorStreamUtil.buildCoordinatorStreamConfig(config) + + + Assert.assertEquals(configMap.get("systems.samzatest.test"), "test") + Assert.assertEquals(configMap.get("test.only"), null) + } + @Test def testReadConfigFromCoordinatorStream { val keyForNonBlankVal = "app.id" From 4bcbd4b71119a5f8a4db55c44cd493764e0d7d39 Mon Sep 17 00:00:00 2001 From: Yuhong Cheng Date: Mon, 23 Sep 2019 13:30:49 -0700 Subject: [PATCH 2/6] address Xinyu's comments --- .../org/apache/samza/config/SystemConfig.java | 1 - ...ava => CoordinatorStreamConfigFactory.java} | 13 ++++++++++++- .../samza/util/CoordinatorStreamUtil.scala | 2 +- .../samza/util/DefaultCoordinatorStream.java | 18 +++++++++--------- .../samza/util/TestCoordinatorStreamUtil.scala | 4 +--- 5 files changed, 23 insertions(+), 15 deletions(-) rename samza-core/src/main/scala/org/apache/samza/util/{CoordinatorStreamUtilFactory.java => CoordinatorStreamConfigFactory.java} (70%) diff --git a/samza-core/src/main/java/org/apache/samza/config/SystemConfig.java b/samza-core/src/main/java/org/apache/samza/config/SystemConfig.java index 52559cb6db..c64a0ea85c 100644 --- a/samza-core/src/main/java/org/apache/samza/config/SystemConfig.java +++ b/samza-core/src/main/java/org/apache/samza/config/SystemConfig.java @@ -37,7 +37,6 @@ public class SystemConfig extends MapConfig { private static final String SYSTEMS_PREFIX = "systems."; public static final String SYSTEM_ID_PREFIX = SYSTEMS_PREFIX + "%s."; - public static final String LI_SYSTEM_ID_PREFIX = SYSTEMS_PREFIX + "li."; private static final String SYSTEM_FACTORY_SUFFIX = ".samza.factory"; diff --git a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtilFactory.java b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamConfigFactory.java similarity index 70% rename from samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtilFactory.java rename to samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamConfigFactory.java index 376b267020..b2b56ae029 100644 --- a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtilFactory.java +++ b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamConfigFactory.java @@ -22,6 +22,17 @@ import org.apache.samza.config.Config; -public interface CoordinatorStreamUtilFactory { +/** + * A CoordinatorStreamConfigFactory receives the job's config and choose specific configs that are needed to + * create coordinator streams. + */ +public interface CoordinatorStreamConfigFactory { + + /** + * Returns a Config what is needed to create coordinator streams. + * + * @param config knonw configs for job + * @return basic configs needed to create coordinator streams + */ Config buildCoordinatorStreamConfig(Config config); } diff --git a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala index bc1d61e653..e3070b6525 100644 --- a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala +++ b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala @@ -43,7 +43,7 @@ object CoordinatorStreamUtil extends Logging { */ def buildCoordinatorStreamConfig(config: Config) = { val buildConfigFactory = config.get(COORDINATOR_STREAM_FACTORY, DEFAULT_COORDINATOR_STREAM) - val coordinatorSystemConfig = Class.forName(buildConfigFactory).newInstance().asInstanceOf[CoordinatorStreamUtilFactory].buildCoordinatorStreamConfig(config) + val coordinatorSystemConfig = Class.forName(buildConfigFactory).newInstance().asInstanceOf[CoordinatorStreamConfigFactory].buildCoordinatorStreamConfig(config) new MapConfig((coordinatorSystemConfig)); diff --git a/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStream.java b/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStream.java index 6073831074..f6f22cc02f 100644 --- a/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStream.java +++ b/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStream.java @@ -27,22 +27,22 @@ import org.apache.samza.config.SystemConfig; -public class DefaultCoordinatorStream implements CoordinatorStreamUtilFactory { +public class DefaultCoordinatorStream implements CoordinatorStreamConfigFactory { @Override public Config buildCoordinatorStreamConfig(Config config) { - Config jobConfig = new JobConfig(config); - String jobName = ((JobConfig) jobConfig).getName().toString(); - String jobId = ((JobConfig) jobConfig).getJobId(); + JobConfig jobConfig = new JobConfig(config); + String jobName = jobConfig.getName().toString(); + String jobId = jobConfig.getJobId(); // Build a map with just the system config and job.name/job.id. This is what's required to start the JobCoordinator. - Map map = config.subset(String.format(SystemConfig.SYSTEM_ID_PREFIX, ((JobConfig) jobConfig).getCoordinatorSystemName()), false); + Map map = config.subset(String.format(SystemConfig.SYSTEM_ID_PREFIX, jobConfig.getCoordinatorSystemName()), false); Map addConfig = new HashMap<>(); addConfig.put(JobConfig.JOB_NAME, jobName); - addConfig.put(((JobConfig) jobConfig).JOB_ID, jobId); - addConfig.put(JobConfig.JOB_COORDINATOR_SYSTEM, ((JobConfig) jobConfig).getCoordinatorSystemName()); - addConfig.put(((JobConfig) jobConfig).MONITOR_PARTITION_CHANGE_FREQUENCY_MS, String.valueOf(((JobConfig) jobConfig).getMonitorPartitionChangeFrequency())); + addConfig.put(jobConfig.JOB_ID, jobId); + addConfig.put(JobConfig.JOB_COORDINATOR_SYSTEM, jobConfig.getCoordinatorSystemName()); + addConfig.put(jobConfig.MONITOR_PARTITION_CHANGE_FREQUENCY_MS, String.valueOf(jobConfig.getMonitorPartitionChangeFrequency())); addConfig.putAll(map); - return new MapConfig(addConfig); + return new MapConfig(addConfig); } } diff --git a/samza-core/src/test/scala/org/apache/samza/util/TestCoordinatorStreamUtil.scala b/samza-core/src/test/scala/org/apache/samza/util/TestCoordinatorStreamUtil.scala index edf9ecb730..dac1fe08a8 100644 --- a/samza-core/src/test/scala/org/apache/samza/util/TestCoordinatorStreamUtil.scala +++ b/samza-core/src/test/scala/org/apache/samza/util/TestCoordinatorStreamUtil.scala @@ -50,9 +50,7 @@ class TestCoordinatorStreamUtil { addConfig.put("systems.samzatest.test","test") addConfig.put("test.only","nothing") val config = new MapConfig(addConfig) - - var configMap = CoordinatorStreamUtil.buildCoordinatorStreamConfig(config) - + val configMap = CoordinatorStreamUtil.buildCoordinatorStreamConfig(config) Assert.assertEquals(configMap.get("systems.samzatest.test"), "test") Assert.assertEquals(configMap.get("test.only"), null) From 6b7fc7876f85f136b5e2b72ca9865341c657d4ec Mon Sep 17 00:00:00 2001 From: Yuhong Cheng Date: Mon, 23 Sep 2019 14:22:36 -0700 Subject: [PATCH 3/6] address Xinyu's comments --- .../apache/samza/util/CoordinatorStreamConfigFactory.java | 4 ++-- .../scala/org/apache/samza/util/CoordinatorStreamUtil.scala | 6 +++--- ...ream.java => DefaultCoordinatorStreamConfigFactory.java} | 2 +- 3 files changed, 6 insertions(+), 6 deletions(-) rename samza-core/src/main/scala/org/apache/samza/util/{DefaultCoordinatorStream.java => DefaultCoordinatorStreamConfigFactory.java} (95%) diff --git a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamConfigFactory.java b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamConfigFactory.java index b2b56ae029..6f930f3867 100644 --- a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamConfigFactory.java +++ b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamConfigFactory.java @@ -23,7 +23,7 @@ /** - * A CoordinatorStreamConfigFactory receives the job's config and choose specific configs that are needed to + * A CoordinatorStreamConfigFactory receives the job's config and create specific configs that are needed to * create coordinator streams. */ public interface CoordinatorStreamConfigFactory { @@ -31,7 +31,7 @@ public interface CoordinatorStreamConfigFactory { /** * Returns a Config what is needed to create coordinator streams. * - * @param config knonw configs for job + * @param config known configs for job * @return basic configs needed to create coordinator streams */ Config buildCoordinatorStreamConfig(Config config); diff --git a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala index e3070b6525..e689726f24 100644 --- a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala +++ b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala @@ -34,15 +34,15 @@ import org.apache.samza.util.ScalaJavaUtil.JavaOptionals import scala.collection.JavaConverters._ object CoordinatorStreamUtil extends Logging { - val COORDINATOR_STREAM_FACTORY = "coordinatorstreamFactory" - val DEFAULT_COORDINATOR_STREAM = "org.apache.samza.util.DefaultCoordinatorStream" + val COORDINATOR_STREAM_FACTORY = "job.coordinatorstream.config.factory" + val DEFAULT_COORDINATOR_STREAM_CONFIG_FACTORY = "org.apache.samza.util.DefaultCoordinatorStreamConfigFactory" /** * Given a job's full config object, build a subset config which includes * only the job name, job id, and system config for the coordinator stream. */ def buildCoordinatorStreamConfig(config: Config) = { - val buildConfigFactory = config.get(COORDINATOR_STREAM_FACTORY, DEFAULT_COORDINATOR_STREAM) + val buildConfigFactory = config.get(COORDINATOR_STREAM_FACTORY, DEFAULT_COORDINATOR_STREAM_CONFIG_FACTORY) val coordinatorSystemConfig = Class.forName(buildConfigFactory).newInstance().asInstanceOf[CoordinatorStreamConfigFactory].buildCoordinatorStreamConfig(config) new MapConfig((coordinatorSystemConfig)); diff --git a/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStream.java b/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStreamConfigFactory.java similarity index 95% rename from samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStream.java rename to samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStreamConfigFactory.java index f6f22cc02f..483a49968d 100644 --- a/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStream.java +++ b/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStreamConfigFactory.java @@ -27,7 +27,7 @@ import org.apache.samza.config.SystemConfig; -public class DefaultCoordinatorStream implements CoordinatorStreamConfigFactory { +public class DefaultCoordinatorStreamConfigFactory implements CoordinatorStreamConfigFactory { @Override public Config buildCoordinatorStreamConfig(Config config) { JobConfig jobConfig = new JobConfig(config); From 864e9a6d2961ae89fb1822ff8018db444fd0ac8b Mon Sep 17 00:00:00 2001 From: Yuhong Cheng Date: Mon, 23 Sep 2019 14:28:56 -0700 Subject: [PATCH 4/6] delete empty line --- .../src/main/java/org/apache/samza/config/SystemConfig.java | 1 - 1 file changed, 1 deletion(-) diff --git a/samza-core/src/main/java/org/apache/samza/config/SystemConfig.java b/samza-core/src/main/java/org/apache/samza/config/SystemConfig.java index c64a0ea85c..93a0c32974 100644 --- a/samza-core/src/main/java/org/apache/samza/config/SystemConfig.java +++ b/samza-core/src/main/java/org/apache/samza/config/SystemConfig.java @@ -38,7 +38,6 @@ public class SystemConfig extends MapConfig { private static final String SYSTEMS_PREFIX = "systems."; public static final String SYSTEM_ID_PREFIX = SYSTEMS_PREFIX + "%s."; - private static final String SYSTEM_FACTORY_SUFFIX = ".samza.factory"; public static final String SYSTEM_FACTORY_FORMAT = SYSTEMS_PREFIX + "%s" + SYSTEM_FACTORY_SUFFIX; @VisibleForTesting From 481528ee3c2bfd36d15ecc71a1e3b26450ba69c3 Mon Sep 17 00:00:00 2001 From: Yuhong Cheng Date: Tue, 24 Sep 2019 13:11:02 -0700 Subject: [PATCH 5/6] added unit test and clean up code --- .../org/apache/samza/config/JobConfig.java | 12 ++++ .../samza/util/CoordinatorStreamUtil.scala | 8 +-- ...DefaultCoordinatorStreamConfigFactory.java | 2 +- ...DefaultCoordinatorStreamConfigFactory.java | 64 +++++++++++++++++++ 4 files changed, 79 insertions(+), 7 deletions(-) create mode 100644 samza-core/src/test/java/org/apache/samza/util/TestDefaultCoordinatorStreamConfigFactory.java diff --git a/samza-core/src/main/java/org/apache/samza/config/JobConfig.java b/samza-core/src/main/java/org/apache/samza/config/JobConfig.java index 9bdcfd5cfa..597109aa28 100644 --- a/samza-core/src/main/java/org/apache/samza/config/JobConfig.java +++ b/samza-core/src/main/java/org/apache/samza/config/JobConfig.java @@ -125,6 +125,9 @@ public class JobConfig extends MapConfig { public static final String CONTAINER_METADATA_FILENAME_FORMAT = "%s.metadata"; // Filename: .metadata public static final String CONTAINER_METADATA_DIRECTORY_SYS_PROPERTY = "samza.log.dir"; + private static final String COORDINATOR_STREAM_FACTORY = "job.coordinatorstream.config.factory"; + private static final String DEFAULT_COORDINATOR_STREAM_CONFIG_FACTORY = "org.apache.samza.util.DefaultCoordinatorStreamConfigFactory"; + public JobConfig(Config config) { super(config); } @@ -341,4 +344,13 @@ public static Optional getMetadataFile(String execEnvContainerId) { new File(dir, String.format(CONTAINER_METADATA_FILENAME_FORMAT, execEnvContainerId))); } } + + /** + * Get coordinatorStreamFactory according to the configs + * @return the name of coordinatorStreamFactory + */ + public static String getCoordinatorStreamFactory(Config config) { + return config.get(COORDINATOR_STREAM_FACTORY, DEFAULT_COORDINATOR_STREAM_CONFIG_FACTORY); + } + } \ No newline at end of file diff --git a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala index e689726f24..029ea8706f 100644 --- a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala +++ b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala @@ -1,5 +1,4 @@ /* - * * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information @@ -34,18 +33,15 @@ import org.apache.samza.util.ScalaJavaUtil.JavaOptionals import scala.collection.JavaConverters._ object CoordinatorStreamUtil extends Logging { - val COORDINATOR_STREAM_FACTORY = "job.coordinatorstream.config.factory" - val DEFAULT_COORDINATOR_STREAM_CONFIG_FACTORY = "org.apache.samza.util.DefaultCoordinatorStreamConfigFactory" - /** * Given a job's full config object, build a subset config which includes * only the job name, job id, and system config for the coordinator stream. */ def buildCoordinatorStreamConfig(config: Config) = { - val buildConfigFactory = config.get(COORDINATOR_STREAM_FACTORY, DEFAULT_COORDINATOR_STREAM_CONFIG_FACTORY) + val buildConfigFactory = JobConfig.getCoordinatorStreamFactory(config); val coordinatorSystemConfig = Class.forName(buildConfigFactory).newInstance().asInstanceOf[CoordinatorStreamConfigFactory].buildCoordinatorStreamConfig(config) - new MapConfig((coordinatorSystemConfig)); + new MapConfig(coordinatorSystemConfig); } diff --git a/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStreamConfigFactory.java b/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStreamConfigFactory.java index 483a49968d..13a7c79658 100644 --- a/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStreamConfigFactory.java +++ b/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStreamConfigFactory.java @@ -31,7 +31,7 @@ public class DefaultCoordinatorStreamConfigFactory implements CoordinatorStreamC @Override public Config buildCoordinatorStreamConfig(Config config) { JobConfig jobConfig = new JobConfig(config); - String jobName = jobConfig.getName().toString(); + String jobName = jobConfig.getName().get(); String jobId = jobConfig.getJobId(); // Build a map with just the system config and job.name/job.id. This is what's required to start the JobCoordinator. diff --git a/samza-core/src/test/java/org/apache/samza/util/TestDefaultCoordinatorStreamConfigFactory.java b/samza-core/src/test/java/org/apache/samza/util/TestDefaultCoordinatorStreamConfigFactory.java new file mode 100644 index 0000000000..aeaaf53e5f --- /dev/null +++ b/samza-core/src/test/java/org/apache/samza/util/TestDefaultCoordinatorStreamConfigFactory.java @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +package org.apache.samza.util; + +import java.util.HashMap; +import java.util.Map; +import java.util.NoSuchElementException; +import org.apache.samza.config.Config; +import org.apache.samza.config.JobConfig; +import org.apache.samza.config.MapConfig; +import org.junit.Test; + +import static org.junit.Assert.*; + + +public class TestDefaultCoordinatorStreamConfigFactory { + + DefaultCoordinatorStreamConfigFactory factory = new DefaultCoordinatorStreamConfigFactory(); + + @Test + public void testBuildCoordinatorStreamConfigWithJobName() { + Map mapConfig = new HashMap<>(); + mapConfig.put("job.name", "testName"); + mapConfig.put("job.id", "testId"); + mapConfig.put("job.coordinator.system", "testSamza"); + mapConfig.put("test.only", "nothing"); + mapConfig.put("systems.testSamza.test", "test"); + + Config config = factory.buildCoordinatorStreamConfig(new MapConfig(mapConfig)); + + assertEquals(config.get("systems.testSamza.test"), "test"); + assertEquals(config.get(JobConfig.JOB_NAME), "testName"); + assertNull(config.get("test.only")); + } + + @Test(expected = NoSuchElementException.class) + public void testBuildCoordinatorStreamConfigWithoutJobName() { + Map mapConfig = new HashMap<>(); + mapConfig.put("job.id", "testId"); + mapConfig.put("job.coordinator.system", "testSamza"); + mapConfig.put("test.only", "nothing"); + mapConfig.put("systems.testSamza.test", "test"); + + factory.buildCoordinatorStreamConfig(new MapConfig(mapConfig)); + } +} From d24a9d6c513fbeb797830f5711755da0f781c1e7 Mon Sep 17 00:00:00 2001 From: Yuhong Cheng Date: Tue, 24 Sep 2019 15:49:58 -0700 Subject: [PATCH 6/6] addressed comments --- .../java/org/apache/samza/config/JobConfig.java | 8 ++++---- .../apache/samza/util/CoordinatorStreamUtil.scala | 3 ++- .../DefaultCoordinatorStreamConfigFactory.java | 8 +++++--- .../org/apache/samza/config/TestJobConfig.java | 9 +++++++++ ...TestDefaultCoordinatorStreamConfigFactory.java | 15 ++++++++++----- 5 files changed, 30 insertions(+), 13 deletions(-) diff --git a/samza-core/src/main/java/org/apache/samza/config/JobConfig.java b/samza-core/src/main/java/org/apache/samza/config/JobConfig.java index 597109aa28..efff5724b6 100644 --- a/samza-core/src/main/java/org/apache/samza/config/JobConfig.java +++ b/samza-core/src/main/java/org/apache/samza/config/JobConfig.java @@ -125,8 +125,8 @@ public class JobConfig extends MapConfig { public static final String CONTAINER_METADATA_FILENAME_FORMAT = "%s.metadata"; // Filename: .metadata public static final String CONTAINER_METADATA_DIRECTORY_SYS_PROPERTY = "samza.log.dir"; - private static final String COORDINATOR_STREAM_FACTORY = "job.coordinatorstream.config.factory"; - private static final String DEFAULT_COORDINATOR_STREAM_CONFIG_FACTORY = "org.apache.samza.util.DefaultCoordinatorStreamConfigFactory"; + public static final String COORDINATOR_STREAM_FACTORY = "job.coordinatorstream.config.factory"; + public static final String DEFAULT_COORDINATOR_STREAM_CONFIG_FACTORY = "org.apache.samza.util.DefaultCoordinatorStreamConfigFactory"; public JobConfig(Config config) { super(config); @@ -349,8 +349,8 @@ public static Optional getMetadataFile(String execEnvContainerId) { * Get coordinatorStreamFactory according to the configs * @return the name of coordinatorStreamFactory */ - public static String getCoordinatorStreamFactory(Config config) { - return config.get(COORDINATOR_STREAM_FACTORY, DEFAULT_COORDINATOR_STREAM_CONFIG_FACTORY); + public String getCoordinatorStreamFactory() { + return get(COORDINATOR_STREAM_FACTORY, DEFAULT_COORDINATOR_STREAM_CONFIG_FACTORY); } } \ No newline at end of file diff --git a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala index 029ea8706f..f6636e7663 100644 --- a/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala +++ b/samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala @@ -38,7 +38,8 @@ object CoordinatorStreamUtil extends Logging { * only the job name, job id, and system config for the coordinator stream. */ def buildCoordinatorStreamConfig(config: Config) = { - val buildConfigFactory = JobConfig.getCoordinatorStreamFactory(config); + val jobConfig = new JobConfig(config) + val buildConfigFactory = jobConfig.getCoordinatorStreamFactory(); val coordinatorSystemConfig = Class.forName(buildConfigFactory).newInstance().asInstanceOf[CoordinatorStreamConfigFactory].buildCoordinatorStreamConfig(config) new MapConfig(coordinatorSystemConfig); diff --git a/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStreamConfigFactory.java b/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStreamConfigFactory.java index 13a7c79658..f3bf10faba 100644 --- a/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStreamConfigFactory.java +++ b/samza-core/src/main/scala/org/apache/samza/util/DefaultCoordinatorStreamConfigFactory.java @@ -22,6 +22,7 @@ import java.util.HashMap; import java.util.Map; import org.apache.samza.config.Config; +import org.apache.samza.config.ConfigException; import org.apache.samza.config.JobConfig; import org.apache.samza.config.MapConfig; import org.apache.samza.config.SystemConfig; @@ -31,16 +32,17 @@ public class DefaultCoordinatorStreamConfigFactory implements CoordinatorStreamC @Override public Config buildCoordinatorStreamConfig(Config config) { JobConfig jobConfig = new JobConfig(config); - String jobName = jobConfig.getName().get(); + String jobName = jobConfig.getName().orElseThrow(() -> new ConfigException("Missing required config: job.name")); + String jobId = jobConfig.getJobId(); // Build a map with just the system config and job.name/job.id. This is what's required to start the JobCoordinator. Map map = config.subset(String.format(SystemConfig.SYSTEM_ID_PREFIX, jobConfig.getCoordinatorSystemName()), false); Map addConfig = new HashMap<>(); addConfig.put(JobConfig.JOB_NAME, jobName); - addConfig.put(jobConfig.JOB_ID, jobId); + addConfig.put(JobConfig.JOB_ID, jobId); addConfig.put(JobConfig.JOB_COORDINATOR_SYSTEM, jobConfig.getCoordinatorSystemName()); - addConfig.put(jobConfig.MONITOR_PARTITION_CHANGE_FREQUENCY_MS, String.valueOf(jobConfig.getMonitorPartitionChangeFrequency())); + addConfig.put(JobConfig.MONITOR_PARTITION_CHANGE_FREQUENCY_MS, String.valueOf(jobConfig.getMonitorPartitionChangeFrequency())); addConfig.putAll(map); return new MapConfig(addConfig); diff --git a/samza-core/src/test/java/org/apache/samza/config/TestJobConfig.java b/samza-core/src/test/java/org/apache/samza/config/TestJobConfig.java index cf14ae3851..dab0f77ef6 100644 --- a/samza-core/src/test/java/org/apache/samza/config/TestJobConfig.java +++ b/samza-core/src/test/java/org/apache/samza/config/TestJobConfig.java @@ -576,4 +576,13 @@ public void testGetMetadataFile() { assertEquals(Optional.empty(), JobConfig.getMetadataFile(null)); } + + @Test + public void testGetCoordinatorStreamFactory() { + JobConfig jobConfig = new JobConfig(new MapConfig(ImmutableMap.of("test", ""))); + assertEquals(jobConfig.getCoordinatorStreamFactory(), JobConfig.DEFAULT_COORDINATOR_STREAM_CONFIG_FACTORY); + + jobConfig = new JobConfig(new MapConfig(ImmutableMap.of(JobConfig.COORDINATOR_STREAM_FACTORY, "specific_coordinator_stream"))); + assertEquals(jobConfig.getCoordinatorStreamFactory(), "specific_coordinator_stream"); + } } \ No newline at end of file diff --git a/samza-core/src/test/java/org/apache/samza/util/TestDefaultCoordinatorStreamConfigFactory.java b/samza-core/src/test/java/org/apache/samza/util/TestDefaultCoordinatorStreamConfigFactory.java index aeaaf53e5f..209dd05a3e 100644 --- a/samza-core/src/test/java/org/apache/samza/util/TestDefaultCoordinatorStreamConfigFactory.java +++ b/samza-core/src/test/java/org/apache/samza/util/TestDefaultCoordinatorStreamConfigFactory.java @@ -22,8 +22,8 @@ import java.util.HashMap; import java.util.Map; -import java.util.NoSuchElementException; import org.apache.samza.config.Config; +import org.apache.samza.config.ConfigException; import org.apache.samza.config.JobConfig; import org.apache.samza.config.MapConfig; import org.junit.Test; @@ -46,12 +46,17 @@ public void testBuildCoordinatorStreamConfigWithJobName() { Config config = factory.buildCoordinatorStreamConfig(new MapConfig(mapConfig)); - assertEquals(config.get("systems.testSamza.test"), "test"); - assertEquals(config.get(JobConfig.JOB_NAME), "testName"); - assertNull(config.get("test.only")); + Map expectedMap = new HashMap<>(); + expectedMap.put("job.name", "testName"); + expectedMap.put("job.id", "testId"); + expectedMap.put("systems.testSamza.test", "test"); + expectedMap.put(JobConfig.JOB_COORDINATOR_SYSTEM, "testSamza"); + expectedMap.put(JobConfig.MONITOR_PARTITION_CHANGE_FREQUENCY_MS, "300000"); + + assertEquals(config, new MapConfig(expectedMap)); } - @Test(expected = NoSuchElementException.class) + @Test(expected = ConfigException.class) public void testBuildCoordinatorStreamConfigWithoutJobName() { Map mapConfig = new HashMap<>(); mapConfig.put("job.id", "testId");