This is an automated email from the ASF dual-hosted git repository.
cameronlee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/samza.git
The following commit(s) were added to refs/heads/master by this push:
new e6e0f46 SAMZA-2429: Update LocalApplicationRunner to load full job
config from config loader when present. (#1255)
e6e0f46 is described below
commit e6e0f46fae68f3315dcffe293b51863a83737606
Author: Ke Wu <[email protected]>
AuthorDate: Thu Jan 30 10:26:03 2020 -0800
SAMZA-2429: Update LocalApplicationRunner to load full job config from
config loader when present. (#1255)
Design:
https://cwiki.apache.org/confluence/display/SAMZA/SEP-23%3A+Simplify+Job+Runner
Changes:
1. Update unit test only public constructor of LocalApplicationRunner to
take SamzaApplication and Config as well to be consistent with other
constructors.
2. Update private constructor of LocalApplicationRunner to take
SamzaApplication and Config.
3. Update private constructor of LocalApplicationRunner to load full job
config when config loader is present.
---
.../samza/runtime/LocalApplicationRunner.java | 62 ++++++++++++++++------
.../samza/runtime/TestLocalApplicationRunner.java | 53 +++++++-----------
2 files changed, 66 insertions(+), 49 deletions(-)
diff --git
a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
index 9ad52b1..8c3c029 100644
---
a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
+++
b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
@@ -62,6 +62,7 @@ import org.apache.samza.system.SystemAdmins;
import org.apache.samza.system.SystemStream;
import org.apache.samza.task.TaskFactory;
import org.apache.samza.task.TaskFactoryUtil;
+import org.apache.samza.util.ConfigUtil;
import org.apache.samza.util.CoordinatorStreamUtil;
import org.apache.samza.util.ReflectionUtil;
import org.apache.samza.zk.ZkJobCoordinatorFactory;
@@ -100,7 +101,7 @@ public class LocalApplicationRunner implements
ApplicationRunner {
* @param config configuration for the application
*/
public LocalApplicationRunner(SamzaApplication app, Config config) {
- this(app, config, getDefaultCoordinatorStreamStoreFactory(new
JobConfig(config)));
+ this(new LocalApplicationRunnerContext(app, config));
}
/**
@@ -111,48 +112,57 @@ public class LocalApplicationRunner implements
ApplicationRunner {
* @param metadataStoreFactory the instance of {@link MetadataStoreFactory}
to read and write to coordinator stream.
*/
public LocalApplicationRunner(SamzaApplication app, Config config,
MetadataStoreFactory metadataStoreFactory) {
- this(ApplicationDescriptorUtil.getAppDescriptor(app, config),
getCoordinationUtils(config), metadataStoreFactory);
+ this(new LocalApplicationRunnerContext(app,
config).setMetadataStoreFactory(metadataStoreFactory));
}
/**
* Constructor only used in unit test to allow injection of {@link
LocalJobPlanner}
*/
@VisibleForTesting
- LocalApplicationRunner(ApplicationDescriptorImpl<? extends
ApplicationDescriptor> appDesc, Optional<CoordinationUtils> coordinationUtils) {
- this(appDesc, coordinationUtils,
getDefaultCoordinatorStreamStoreFactory(new JobConfig(appDesc.getConfig())));
+ LocalApplicationRunner(SamzaApplication app, Config config,
CoordinationUtils coordinationUtils) {
+ this(new LocalApplicationRunnerContext(app,
config).setCoordinationUtils(coordinationUtils));
}
- private LocalApplicationRunner(
- ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc,
- Optional<CoordinationUtils> coordinationUtils,
- MetadataStoreFactory metadataStoreFactory) {
- this.appDesc = appDesc;
- this.isAppModeBatch = isAppModeBatch(appDesc.getConfig());
- this.coordinationUtils = coordinationUtils;
- this.metadataStoreFactory = Optional.ofNullable(metadataStoreFactory);
+ private LocalApplicationRunner(LocalApplicationRunnerContext context) {
+ Config config = context.config;
+ if (new JobConfig(context.config).getConfigLoaderFactory().isPresent()) {
+ config = ConfigUtil.loadConfig(config);
+ }
+
+ this.appDesc = ApplicationDescriptorUtil.getAppDescriptor(context.app,
config);
+ this.isAppModeBatch = isAppModeBatch(config);
+ this.coordinationUtils = context.coordinationUtils.isPresent()
+ ? context.coordinationUtils
+ : getCoordinationUtils(config);
+ this.metadataStoreFactory = context.metadataStoreFactory.isPresent()
+ ? context.metadataStoreFactory
+ : getDefaultCoordinatorStreamStoreFactory(config);
}
@VisibleForTesting
- static MetadataStoreFactory
getDefaultCoordinatorStreamStoreFactory(JobConfig jobConfig) {
+ static Optional<MetadataStoreFactory>
getDefaultCoordinatorStreamStoreFactory(Config config) {
+ JobConfig jobConfig = new JobConfig(config);
+
String coordinatorSystemName = jobConfig.getCoordinatorSystemNameOrNull();
JobCoordinatorConfig jobCoordinatorConfig = new
JobCoordinatorConfig(jobConfig);
String jobCoordinatorFactoryClassName =
jobCoordinatorConfig.getJobCoordinatorFactoryClassName();
// TODO: Remove restriction to only ZkJobCoordinator after next phase of
metadata store abstraction.
if (StringUtils.isNotBlank(coordinatorSystemName) &&
ZkJobCoordinatorFactory.class.getName().equals(jobCoordinatorFactoryClassName))
{
- return new CoordinatorStreamMetadataStoreFactory();
+ return Optional.of(new CoordinatorStreamMetadataStoreFactory());
}
LOG.warn("{} or {} not configured, or {} is not {}. No default coordinator
stream metadata store will be created.",
JobConfig.JOB_COORDINATOR_SYSTEM, JobConfig.JOB_DEFAULT_SYSTEM,
JobCoordinatorConfig.JOB_COORDINATOR_FACTORY,
ZkJobCoordinatorFactory.class.getName());
- return null;
+ return Optional.empty();
}
private static Optional<CoordinationUtils> getCoordinationUtils(Config
config) {
if (!isAppModeBatch(config)) {
return Optional.empty();
}
+
JobCoordinatorConfig jcConfig = new JobCoordinatorConfig(config);
CoordinationUtils coordinationUtils =
jcConfig.getCoordinationUtilsFactory()
.getCoordinationUtils(CoordinationConstants.APPLICATION_RUNNER_PATH_SUFFIX,
PROCESSOR_ID, config);
@@ -479,4 +489,26 @@ public class LocalApplicationRunner implements
ApplicationRunner {
}
}
}
+
+ private final static class LocalApplicationRunnerContext {
+ SamzaApplication app;
+ Config config;
+ Optional<CoordinationUtils> coordinationUtils = Optional.empty();
+ Optional<MetadataStoreFactory> metadataStoreFactory = Optional.empty();
+
+ LocalApplicationRunnerContext(SamzaApplication app, Config config) {
+ this.app = app;
+ this.config = config;
+ }
+
+ LocalApplicationRunnerContext setCoordinationUtils(CoordinationUtils
coordinationUtils) {
+ this.coordinationUtils = Optional.of(coordinationUtils);
+ return this;
+ }
+
+ LocalApplicationRunnerContext setMetadataStoreFactory(MetadataStoreFactory
metadataStoreFactory) {
+ this.metadataStoreFactory = Optional.of(metadataStoreFactory);
+ return this;
+ }
+ }
}
diff --git
a/samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java
b/samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java
index 6c7fcb4..972a042 100644
---
a/samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java
+++
b/samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java
@@ -83,7 +83,6 @@ import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-
@RunWith(PowerMockRunner.class)
@PrepareForTest({LocalJobPlanner.class, LocalApplicationRunner.class,
ZkMetadataStoreFactory.class})
public class TestLocalApplicationRunner {
@@ -185,9 +184,7 @@ public class TestLocalApplicationRunner {
cfgs.put(ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS,
UUIDGenerator.class.getName());
config = new MapConfig(cfgs);
ProcessorLifecycleListenerFactory mockFactory = (pContext, cfg) ->
mock(ProcessorLifecycleListener.class);
- mockApp = (StreamApplication) appDesc -> {
- appDesc.withProcessorLifecycleListenerFactory(mockFactory);
- };
+ mockApp = (StreamApplication) appDesc ->
appDesc.withProcessorLifecycleListenerFactory(mockFactory);
prepareTest();
// return the jobConfigs from the planner
@@ -226,9 +223,7 @@ public class TestLocalApplicationRunner {
cfgs.put(ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS,
UUIDGenerator.class.getName());
config = new MapConfig(cfgs);
ProcessorLifecycleListenerFactory mockFactory = (pContext, cfg) ->
mock(ProcessorLifecycleListener.class);
- mockApp = (StreamApplication) appDesc -> {
- appDesc.withProcessorLifecycleListenerFactory(mockFactory);
- };
+ mockApp = (StreamApplication) appDesc ->
appDesc.withProcessorLifecycleListenerFactory(mockFactory);
prepareTest();
// return the jobConfigs from the planner
@@ -263,9 +258,7 @@ public class TestLocalApplicationRunner {
cfgs.put(ApplicationConfig.PROCESSOR_ID, "0");
config = new MapConfig(cfgs);
ProcessorLifecycleListenerFactory mockFactory = (pContext, cfg) ->
mock(ProcessorLifecycleListener.class);
- mockApp = (StreamApplication) appDesc -> {
- appDesc.withProcessorLifecycleListenerFactory(mockFactory);
- };
+ mockApp = (StreamApplication) appDesc ->
appDesc.withProcessorLifecycleListenerFactory(mockFactory);
prepareTest();
// return the jobConfigs from the planner
@@ -306,9 +299,7 @@ public class TestLocalApplicationRunner {
cfgs.put(ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS,
UUIDGenerator.class.getName());
config = new MapConfig(cfgs);
ProcessorLifecycleListenerFactory mockFactory = (pContext, cfg) ->
mock(ProcessorLifecycleListener.class);
- mockApp = (StreamApplication) appDesc -> {
- appDesc.withProcessorLifecycleListenerFactory(mockFactory);
- };
+ mockApp = (StreamApplication) appDesc ->
appDesc.withProcessorLifecycleListenerFactory(mockFactory);
prepareTest();
// return the jobConfigs from the planner
@@ -353,9 +344,7 @@ public class TestLocalApplicationRunner {
cfgs.put(ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS,
UUIDGenerator.class.getName());
config = new MapConfig(cfgs);
ProcessorLifecycleListenerFactory mockFactory = (pContext, cfg) ->
mock(ProcessorLifecycleListener.class);
- mockApp = (StreamApplication) appDesc -> {
- appDesc.withProcessorLifecycleListenerFactory(mockFactory);
- };
+ mockApp = (StreamApplication) appDesc ->
appDesc.withProcessorLifecycleListenerFactory(mockFactory);
prepareTest();
// return the jobConfigs from the planner
@@ -443,7 +432,7 @@ public class TestLocalApplicationRunner {
ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc =
ApplicationDescriptorUtil.getAppDescriptor(mockApp, config);
localPlanner = spy(new LocalJobPlanner(appDesc, coordinationUtils,
"FAKE_UID", "FAKE_RUNID"));
- runner = spy(new LocalApplicationRunner(appDesc,
Optional.of(coordinationUtils)));
+ runner = spy(new LocalApplicationRunner(mockApp, config,
coordinationUtils));
doReturn(localPlanner).when(runner).getPlanner();
}
@@ -451,7 +440,6 @@ public class TestLocalApplicationRunner {
* For app.mode=BATCH ensure that the run.id generation utils --
* DistributedLock, ClusterMembership and MetadataStore are created.
* Also ensure that metadataStore.put is invoked (to write the run.id)
- * @throws Exception
*/
@Test
public void testRunIdForBatch() throws Exception {
@@ -475,7 +463,6 @@ public class TestLocalApplicationRunner {
* For app.mode=STREAM ensure that the run.id generation utils --
* DistributedLock, ClusterMembership and MetadataStore are NOT created.
* Also ensure that metadataStore.put is NOT invoked
- * @throws Exception
*/
@Test
public void testRunIdForStream() throws Exception {
@@ -515,7 +502,7 @@ public class TestLocalApplicationRunner {
ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc =
ApplicationDescriptorUtil.getAppDescriptor(mockApp, config);
- runner = spy(new LocalApplicationRunner(appDesc,
Optional.of(coordinationUtils)));
+ runner = spy(new LocalApplicationRunner(mockApp, config,
coordinationUtils));
localPlanner = spy(new LocalJobPlanner(appDesc, coordinationUtils,
"FAKE_UID", "FAKE_RUNID"));
doReturn(localPlanner).when(runner).getPlanner();
StreamProcessor sp = mock(StreamProcessor.class);
@@ -531,9 +518,9 @@ public class TestLocalApplicationRunner {
*/
@Test
public void
testGetCoordinatorStreamStoreFactoryWithoutJobCoordinatorSystem() {
- MetadataStoreFactory metadataStoreFactory =
- LocalApplicationRunner.getDefaultCoordinatorStreamStoreFactory(new
JobConfig(new MapConfig()));
- assertNull(metadataStoreFactory);
+ Optional<MetadataStoreFactory> metadataStoreFactory =
+ LocalApplicationRunner.getDefaultCoordinatorStreamStoreFactory(new
MapConfig());
+ assertFalse(metadataStoreFactory.isPresent());
}
/**
@@ -542,9 +529,9 @@ public class TestLocalApplicationRunner {
*/
@Test
public void testGetCoordinatorStreamStoreFactoryWithJobCoordinatorSystem() {
- MetadataStoreFactory metadataStoreFactory =
- LocalApplicationRunner.getDefaultCoordinatorStreamStoreFactory(new
JobConfig(new MapConfig(ImmutableMap.of(JobConfig.JOB_COORDINATOR_SYSTEM,
"test-system"))));
- assertNotNull(metadataStoreFactory);
+ Optional<MetadataStoreFactory> metadataStoreFactory =
+ LocalApplicationRunner.getDefaultCoordinatorStreamStoreFactory(new
MapConfig(ImmutableMap.of(JobConfig.JOB_COORDINATOR_SYSTEM, "test-system")));
+ assertTrue(metadataStoreFactory.isPresent());
}
/**
@@ -553,9 +540,9 @@ public class TestLocalApplicationRunner {
*/
@Test
public void testGetCoordinatorStreamStoreFactoryWithDefaultSystem() {
- MetadataStoreFactory metadataStoreFactory =
- LocalApplicationRunner.getDefaultCoordinatorStreamStoreFactory(new
JobConfig(new MapConfig(ImmutableMap.of(JobConfig.JOB_DEFAULT_SYSTEM,
"test-system"))));
- assertNotNull(metadataStoreFactory);
+ Optional<MetadataStoreFactory> metadataStoreFactory =
+ LocalApplicationRunner.getDefaultCoordinatorStreamStoreFactory(new
MapConfig(ImmutableMap.of(JobConfig.JOB_DEFAULT_SYSTEM, "test-system")));
+ assertTrue(metadataStoreFactory.isPresent());
}
/**
@@ -568,14 +555,13 @@ public class TestLocalApplicationRunner {
ImmutableMap.of(
JobConfig.JOB_DEFAULT_SYSTEM, "test-system",
JobCoordinatorConfig.JOB_COORDINATOR_FACTORY,
PassthroughJobCoordinatorFactory.class.getName()));
- MetadataStoreFactory metadataStoreFactory =
- LocalApplicationRunner.getDefaultCoordinatorStreamStoreFactory(new
JobConfig(mapConfig));
- assertNull(metadataStoreFactory);
+ Optional<MetadataStoreFactory> metadataStoreFactory =
+
LocalApplicationRunner.getDefaultCoordinatorStreamStoreFactory(mapConfig);
+ assertFalse(metadataStoreFactory.isPresent());
}
/**
* Underlying coordinator stream should be created if using
CoordinatorStreamMetadataStoreFactory
- * @throws Exception
*/
@Test
public void testCreateCoordinatorStreamWithCoordinatorFactory() throws
Exception {
@@ -600,7 +586,6 @@ public class TestLocalApplicationRunner {
/**
* Underlying coordinator stream should not be created if not using
CoordinatorStreamMetadataStoreFactory
- * @throws Exception
*/
@Test
public void testCreateCoordinatorStreamWithoutCoordinatorFactory() throws
Exception {
