You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ja...@apache.org on 2018/12/07 02:11:45 UTC
samza git commit: SAMZA-1835: Consolidate all processorId generation
code.
Repository: samza
Updated Branches:
refs/heads/master bd9387b7f -> df56f2dc1
SAMZA-1835: Consolidate all processorId generation code.
Currently, the processorId creation function createProcessorId() is repeated in three different implementation of `JobCoordinator` viz `ZkJobCoordinator`, `PassthroughJobCoordinator`, and `AzureJobCoordinator`. Here're the few problems that stems from this duplication.
1. `ProcessorId` is passed into the `MetricsReporterFactory` through the factory create method: `MetricsReporter getMetricsReporter(String name, String processorId, Config config);`. Custom `MetricsReporter` implementations currently use the processorId as a component in the generated metric names. Metrics reporters are instantiated from `LocalApplicationRunner` and`processorId` is currently passed in as null to `MetricsReporterFactory.getMetricsReporter`. This corrupts the generated metrics names.
2. `ZkJobCoordinator`, `ZkUtils`, `ZkLeaderElector` and different downstream components of `LocalApplicationRunner` currently instantiate and manage their private reporters, rather than the sharing common `MetricsRegistry` managed by `LocalApplicationRunner`. Since there is no common namespace and reporter shared by downstream components of `LocalApplicationRunner`, generating metrics dashboards for standalone is kind of a hassle.
This PR is comprised of the following changes:
1. Moved the processorId generation to `LocalApplicationRunner` and injects the generated `processorId` to all the downstream layers.
2. Deprecated the getProcessorId API in `JobCoordinator` interface.
3. Add the `processorId` and `metricsRegistry` arguments to the `getJobCoordinator` method of `JobCoordinatorFactory` interface.
4. Fixed the unit tests and added unit tests for `LocalApplicationRunner.createProcessorId`.
Author: Shanthoosh Venkataraman <sv...@linkedin.com>
Author: Shanthoosh Venkataraman <sp...@usc.edu>
Author: svenkata <sv...@linkedin.com>
Reviewers: Jagadish<ja...@apache.org>
Closes #844 from shanthoosh/SAMZA-1835
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/df56f2dc
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/df56f2dc
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/df56f2dc
Branch: refs/heads/master
Commit: df56f2dc1ee1be9b2a3aa9894465e3f870451125
Parents: bd9387b
Author: Shanthoosh Venkataraman <sv...@linkedin.com>
Authored: Thu Dec 6 18:11:38 2018 -0800
Committer: Jagadish <jv...@linkedin.com>
Committed: Thu Dec 6 18:11:38 2018 -0800
----------------------------------------------------------------------
.../samza/coordinator/AzureJobCoordinator.java | 25 ++---------
.../coordinator/AzureJobCoordinatorFactory.java | 5 ++-
.../apache/samza/config/ApplicationConfig.java | 1 -
.../samza/coordinator/JobCoordinator.java | 1 +
.../coordinator/JobCoordinatorFactory.java | 11 +++--
.../apache/samza/processor/StreamProcessor.java | 45 ++++++++++++-------
.../samza/runtime/LocalApplicationRunner.java | 32 +++++++++++--
.../standalone/PassthroughJobCoordinator.java | 25 ++---------
.../PassthroughJobCoordinatorFactory.java | 5 ++-
.../org/apache/samza/zk/ZkJobCoordinator.java | 47 +-------------------
.../samza/zk/ZkJobCoordinatorFactory.java | 6 +--
.../samza/processor/TestStreamProcessor.java | 14 +++---
.../samza/runtime/MockProcessorIdGenerator.java | 29 ++++++++++++
.../runtime/TestLocalApplicationRunner.java | 26 +++++++++++
.../apache/samza/zk/TestZkJobCoordinator.java | 10 ++---
.../processor/TestZkStreamProcessorBase.java | 5 ++-
.../test/processor/TestStreamProcessor.java | 9 ++--
17 files changed, 158 insertions(+), 138 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinator.java
----------------------------------------------------------------------
diff --git a/samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinator.java b/samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinator.java
index 076ab54..ae4aba3 100644
--- a/samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinator.java
+++ b/samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinator.java
@@ -19,12 +19,9 @@
package org.apache.samza.coordinator;
-import org.apache.commons.lang3.StringUtils;
import org.apache.samza.AzureClient;
-import org.apache.samza.config.ApplicationConfig;
import org.apache.samza.config.AzureConfig;
import org.apache.samza.config.Config;
-import org.apache.samza.config.ConfigException;
import org.apache.samza.config.JobConfig;
import org.apache.samza.config.TaskConfig;
import org.apache.samza.container.TaskName;
@@ -42,7 +39,7 @@ import org.apache.samza.coordinator.scheduler.LivenessCheckScheduler;
import org.apache.samza.coordinator.scheduler.RenewLeaseScheduler;
import org.apache.samza.coordinator.scheduler.SchedulerStateChangeListener;
import org.apache.samza.job.model.JobModel;
-import org.apache.samza.runtime.ProcessorIdGenerator;
+import org.apache.samza.metrics.MetricsRegistry;
import org.apache.samza.system.StreamMetadataCache;
import org.apache.samza.system.SystemAdmins;
import org.apache.samza.system.SystemStream;
@@ -101,10 +98,10 @@ public class AzureJobCoordinator implements JobCoordinator {
* Creates an instance of Azure job coordinator, along with references to Azure leader elector, Azure Blob and Azure Table.
* @param config User defined config
*/
- public AzureJobCoordinator(Config config) {
+ public AzureJobCoordinator(String processorId, Config config, MetricsRegistry metricsRegistry) {
//TODO: Cleanup previous values in the table when barrier times out.
+ this.processorId = processorId;
this.config = config;
- processorId = createProcessorId(config);
currentJMVersion = new AtomicReference<>(INITIAL_STATE);
AzureConfig azureConfig = new AzureConfig(config);
AzureClient client = new AzureClient(azureConfig.getAzureConnectionString());
@@ -473,22 +470,6 @@ public class AzureJobCoordinator implements JobCoordinator {
}
}
- private String createProcessorId(Config config) {
- // TODO: This check to be removed after 0.13+
- ApplicationConfig appConfig = new ApplicationConfig(config);
- if (appConfig.getProcessorId() != null) {
- return appConfig.getProcessorId();
- } else if (StringUtils.isNotBlank(appConfig.getAppProcessorIdGeneratorClass())) {
- ProcessorIdGenerator idGenerator =
- Util.getObj(appConfig.getAppProcessorIdGeneratorClass(), ProcessorIdGenerator.class);
- return idGenerator.generateProcessorId(config);
- } else {
- throw new ConfigException(String
- .format("Expected either %s or %s to be configured", ApplicationConfig.PROCESSOR_ID,
- ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS));
- }
- }
-
public class AzureLeaderElectorListener implements LeaderElectorListener {
/**
* Keep renewing the lease and do the required tasks as a leader.
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinatorFactory.java
----------------------------------------------------------------------
diff --git a/samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinatorFactory.java b/samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinatorFactory.java
index 8b3d357..ff8925a 100644
--- a/samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinatorFactory.java
+++ b/samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinatorFactory.java
@@ -20,10 +20,11 @@
package org.apache.samza.coordinator;
import org.apache.samza.config.Config;
+import org.apache.samza.metrics.MetricsRegistry;
public class AzureJobCoordinatorFactory implements JobCoordinatorFactory {
@Override
- public JobCoordinator getJobCoordinator(Config config) {
- return new AzureJobCoordinator(config);
+ public JobCoordinator getJobCoordinator(String processorId, Config config, MetricsRegistry metricsRegistry) {
+ return new AzureJobCoordinator(processorId, config, metricsRegistry);
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-core/src/main/java/org/apache/samza/config/ApplicationConfig.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/config/ApplicationConfig.java b/samza-core/src/main/java/org/apache/samza/config/ApplicationConfig.java
index 39facb6..804035e 100644
--- a/samza-core/src/main/java/org/apache/samza/config/ApplicationConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/ApplicationConfig.java
@@ -85,7 +85,6 @@ public class ApplicationConfig extends MapConfig {
return String.format("app-%s-%s", getAppName(), getAppId());
}
- @Deprecated
public String getProcessorId() {
return get(PROCESSOR_ID, null);
}
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinator.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinator.java b/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinator.java
index bd06039..cd10acb 100644
--- a/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinator.java
+++ b/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinator.java
@@ -70,6 +70,7 @@ public interface JobCoordinator {
*
* @return String representing a unique logical processor ID
*/
+ @Deprecated
String getProcessorId();
/**
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinatorFactory.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinatorFactory.java b/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinatorFactory.java
index 83ebf52..8f3d96e 100644
--- a/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinatorFactory.java
+++ b/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinatorFactory.java
@@ -20,13 +20,16 @@ package org.apache.samza.coordinator;
import org.apache.samza.annotation.InterfaceStability;
import org.apache.samza.config.Config;
+import org.apache.samza.metrics.MetricsRegistry;
@InterfaceStability.Evolving
public interface JobCoordinatorFactory {
/**
- * Return a new instance of {@link JobCoordinator}
- * @param config Configs relevant for the JobCoordinator TODO: Separate JC related configs into a "JobCoordinatorConfig"
- * @return {@link JobCoordinator} instance
+ * Returns a new instance of {@link JobCoordinator}.
+ * @param processorId a unique logical identifier assigned to the {@link org.apache.samza.processor.StreamProcessor}.
+ * @param config the configuration of the samza application.
+ * @param metricsRegistry used to publish the coordination specific metrics.
+ * @return the {@link JobCoordinator} instance.
*/
- JobCoordinator getJobCoordinator(Config config);
+ JobCoordinator getJobCoordinator(String processorId, Config config, MetricsRegistry metricsRegistry);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java b/samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java
index 389dafd..2c0c0b7 100644
--- a/samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java
+++ b/samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java
@@ -30,6 +30,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import org.apache.commons.lang3.StringUtils;
import org.apache.samza.annotation.InterfaceStability;
import org.apache.samza.config.Config;
import org.apache.samza.config.JobCoordinatorConfig;
@@ -46,6 +47,7 @@ import org.apache.samza.coordinator.JobCoordinator;
import org.apache.samza.coordinator.JobCoordinatorFactory;
import org.apache.samza.coordinator.JobCoordinatorListener;
import org.apache.samza.job.model.JobModel;
+import org.apache.samza.metrics.MetricsRegistryMap;
import org.apache.samza.metrics.MetricsReporter;
import org.apache.samza.runtime.ProcessorLifecycleListener;
import org.apache.samza.task.TaskFactory;
@@ -123,6 +125,7 @@ public class StreamProcessor {
private final String processorId;
private final ExecutorService containerExcecutorService;
private final Object lock = new Object();
+ private final MetricsRegistryMap metricsRegistry;
private volatile Throwable containerException = null;
@@ -163,25 +166,26 @@ public class StreamProcessor {
JobCoordinatorListener jobCoordinatorListener = null;
/**
- * Same as {@link #StreamProcessor(Config, Map, TaskFactory, ProcessorLifecycleListener, JobCoordinator)}, except
+ * Same as {@link #StreamProcessor(String, Config, Map, TaskFactory, ProcessorLifecycleListener, JobCoordinator)}, except
* it creates a {@link JobCoordinator} instead of accepting it as an argument.
*
- * @param config configuration required to launch {@link JobCoordinator} and {@link SamzaContainer}
- * @param customMetricsReporters registered with the metrics system to report metrics
- * @param taskFactory task factory to instantiate the Task
- * @param processorListener listener to the StreamProcessor life cycle
+ * @param processorId a unique logical identifier assigned to the stream processor.
+ * @param config configuration required to launch {@link JobCoordinator} and {@link SamzaContainer}.
+ * @param customMetricsReporters registered with the metrics system to report metrics.
+ * @param taskFactory the task factory to instantiate the Task.
+ * @param processorListener listener to the StreamProcessor life cycle.
*
- * Deprecated: Use {@link #StreamProcessor(Config, Map, TaskFactory, Optional, Optional, Optional,
+ * Deprecated: Use {@link #StreamProcessor(String, Config, Map, TaskFactory, Optional, Optional, Optional,
* StreamProcessorLifecycleListenerFactory, JobCoordinator)} instead.
*/
@Deprecated
- public StreamProcessor(Config config, Map<String, MetricsReporter> customMetricsReporters, TaskFactory taskFactory,
+ public StreamProcessor(String processorId, Config config, Map<String, MetricsReporter> customMetricsReporters, TaskFactory taskFactory,
ProcessorLifecycleListener processorListener) {
- this(config, customMetricsReporters, taskFactory, processorListener, null);
+ this(processorId, config, customMetricsReporters, taskFactory, processorListener, null);
}
/**
- * Same as {@link #StreamProcessor(Config, Map, TaskFactory, Optional, Optional, Optional,
+ * Same as {@link #StreamProcessor(String, Config, Map, TaskFactory, Optional, Optional, Optional,
* StreamProcessorLifecycleListenerFactory, JobCoordinator)}, with the following differences:
* <ol>
* <li>Passes null for application-defined context factories</li>
@@ -189,25 +193,27 @@ public class StreamProcessor {
* {@link StreamProcessorLifecycleListenerFactory}</li>
* </ol>
*
+ * @param processorId a unique logical identifier assigned to the stream processor.
* @param config configuration required to launch {@link JobCoordinator} and {@link SamzaContainer}
* @param customMetricsReporters registered with the metrics system to report metrics
* @param taskFactory task factory to instantiate the Task
* @param processorListener listener to the StreamProcessor life cycle
* @param jobCoordinator the instance of {@link JobCoordinator}
*
- * Deprecated: Use {@link #StreamProcessor(Config, Map, TaskFactory, Optional, Optional, Optional,
+ * Deprecated: Use {@link #StreamProcessor(String, Config, Map, TaskFactory, Optional, Optional, Optional,
* StreamProcessorLifecycleListenerFactory, JobCoordinator)} instead.
*/
@Deprecated
- public StreamProcessor(Config config, Map<String, MetricsReporter> customMetricsReporters, TaskFactory taskFactory,
+ public StreamProcessor(String processorId, Config config, Map<String, MetricsReporter> customMetricsReporters, TaskFactory taskFactory,
ProcessorLifecycleListener processorListener, JobCoordinator jobCoordinator) {
- this(config, customMetricsReporters, taskFactory, Optional.empty(), Optional.empty(), Optional.empty(),
- sp -> processorListener, jobCoordinator);
+ this(processorId, config, customMetricsReporters, taskFactory, Optional.empty(), Optional.empty(), Optional.empty(), sp -> processorListener,
+ jobCoordinator);
}
/**
* Builds a {@link StreamProcessor} with full specification of processing components.
*
+ * @param processorId a unique logical identifier assigned to the stream processor.
* @param config configuration required to launch {@link JobCoordinator} and {@link SamzaContainer}
* @param customMetricsReporters registered with the metrics system to report metrics
* @param taskFactory task factory to instantiate the Task
@@ -217,14 +223,20 @@ public class StreamProcessor {
* @param listenerFactory factory for creating a listener to the StreamProcessor life cycle
* @param jobCoordinator the instance of {@link JobCoordinator}
*/
- public StreamProcessor(Config config, Map<String, MetricsReporter> customMetricsReporters, TaskFactory taskFactory,
+ public StreamProcessor(String processorId, Config config, Map<String, MetricsReporter> customMetricsReporters, TaskFactory taskFactory,
Optional<ApplicationContainerContextFactory<ApplicationContainerContext>> applicationDefinedContainerContextFactoryOptional,
Optional<ApplicationTaskContextFactory<ApplicationTaskContext>> applicationDefinedTaskContextFactoryOptional,
Optional<ExternalContext> externalContextOptional, StreamProcessorLifecycleListenerFactory listenerFactory,
JobCoordinator jobCoordinator) {
Preconditions.checkNotNull(listenerFactory, "StreamProcessorListenerFactory cannot be null.");
+ Preconditions.checkArgument(StringUtils.isNotBlank(processorId), "ProcessorId cannot be null.");
this.config = config;
+ this.processorId = processorId;
+ this.metricsRegistry = new MetricsRegistryMap();
this.customMetricsReporter = customMetricsReporters;
+ for (MetricsReporter metricsReporter : customMetricsReporter.values()) {
+ metricsReporter.register("StreamProcessor", metricsRegistry);
+ }
this.taskFactory = taskFactory;
this.applicationDefinedContainerContextFactoryOptional = applicationDefinedContainerContextFactoryOptional;
this.applicationDefinedTaskContextFactoryOptional = applicationDefinedTaskContextFactoryOptional;
@@ -235,8 +247,6 @@ public class StreamProcessor {
this.jobCoordinator.setListener(jobCoordinatorListener);
ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat(CONTAINER_THREAD_NAME_FORMAT).setDaemon(true).build();
this.containerExcecutorService = Executors.newSingleThreadExecutor(threadFactory);
- // TODO: remove the dependency on jobCoordinator for processorId after fixing SAMZA-1835
- this.processorId = this.jobCoordinator.getProcessorId();
this.processorListener = listenerFactory.createInstance(this);
}
@@ -287,6 +297,7 @@ public class StreamProcessor {
*/
public void stop() {
synchronized (lock) {
+ LOGGER.info("Stopping the stream processor: {}.", processorId);
if (state != State.STOPPING && state != State.STOPPED) {
state = State.STOPPING;
try {
@@ -328,7 +339,7 @@ public class StreamProcessor {
private JobCoordinator createJobCoordinator() {
String jobCoordinatorFactoryClassName = new JobCoordinatorConfig(config).getJobCoordinatorFactoryClassName();
- return Util.getObj(jobCoordinatorFactoryClassName, JobCoordinatorFactory.class).getJobCoordinator(config);
+ return Util.getObj(jobCoordinatorFactoryClassName, JobCoordinatorFactory.class).getJobCoordinator(processorId, config, metricsRegistry);
}
/**
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
----------------------------------------------------------------------
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 189fc1f..7da3369 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
@@ -32,6 +32,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.commons.lang3.StringUtils;
import org.apache.samza.SamzaException;
import org.apache.samza.application.descriptors.ApplicationDescriptor;
import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
@@ -39,6 +40,7 @@ import org.apache.samza.application.descriptors.ApplicationDescriptorUtil;
import org.apache.samza.application.SamzaApplication;
import org.apache.samza.config.ApplicationConfig;
import org.apache.samza.config.Config;
+import org.apache.samza.config.ConfigException;
import org.apache.samza.config.JobConfig;
import org.apache.samza.context.ExternalContext;
import org.apache.samza.execution.LocalJobPlanner;
@@ -47,6 +49,7 @@ import org.apache.samza.metrics.MetricsReporter;
import org.apache.samza.processor.StreamProcessor;
import org.apache.samza.task.TaskFactory;
import org.apache.samza.task.TaskFactoryUtil;
+import org.apache.samza.util.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -167,14 +170,37 @@ public class LocalApplicationRunner implements ApplicationRunner {
Optional<ExternalContext> externalContextOptional) {
TaskFactory taskFactory = TaskFactoryUtil.getTaskFactory(appDesc);
Map<String, MetricsReporter> reporters = new HashMap<>();
- // TODO: the null processorId has to be fixed after SAMZA-1835
+ String processorId = createProcessorId(new ApplicationConfig(config));
appDesc.getMetricsReporterFactories().forEach((name, factory) ->
- reporters.put(name, factory.getMetricsReporter(name, null, config)));
- return new StreamProcessor(config, reporters, taskFactory, appDesc.getApplicationContainerContextFactory(),
+ reporters.put(name, factory.getMetricsReporter(name, processorId, config)));
+ return new StreamProcessor(processorId, config, reporters, taskFactory, appDesc.getApplicationContainerContextFactory(),
appDesc.getApplicationTaskContextFactory(), externalContextOptional, listenerFactory, null);
}
/**
+ * Generates a unique logical identifier for the stream processor using the provided {@param appConfig}.
+ * 1. If the processorId is defined in the configuration, then returns the value defined in the configuration.
+ * 2. Else if the {@linkplain ProcessorIdGenerator} class is defined the configuration, then uses the {@linkplain ProcessorIdGenerator}
+ * to generate the unique processorId.
+ * 3. Else throws the {@see ConfigException} back to the caller.
+ * @param appConfig the configuration of the samza application.
+ * @throws ConfigException if neither processor.id nor app.processor-id-generator.class is defined in the configuration.
+ * @return the generated processor identifier.
+ */
+ @VisibleForTesting
+ static String createProcessorId(ApplicationConfig appConfig) {
+ if (StringUtils.isNotBlank(appConfig.getProcessorId())) {
+ return appConfig.getProcessorId();
+ } else if (StringUtils.isNotBlank(appConfig.getAppProcessorIdGeneratorClass())) {
+ ProcessorIdGenerator idGenerator = Util.getObj(appConfig.getAppProcessorIdGeneratorClass(), ProcessorIdGenerator.class);
+ return idGenerator.generateProcessorId(appConfig);
+ } else {
+ throw new ConfigException(String.format("Expected either %s or %s to be configured", ApplicationConfig.PROCESSOR_ID,
+ ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS));
+ }
+ }
+
+ /**
* Defines a specific implementation of {@link ProcessorLifecycleListener} for local {@link StreamProcessor}s.
*/
private final class LocalStreamProcessorLifecycleListener implements ProcessorLifecycleListener {
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinator.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinator.java b/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinator.java
index 44fd811..15a9205 100644
--- a/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinator.java
+++ b/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinator.java
@@ -20,9 +20,7 @@ package org.apache.samza.standalone;
import com.google.common.collect.ImmutableMap;
import org.apache.samza.checkpoint.CheckpointManager;
-import org.apache.samza.config.ApplicationConfig;
import org.apache.samza.config.Config;
-import org.apache.samza.config.ConfigException;
import org.apache.samza.config.JobConfig;
import org.apache.samza.config.TaskConfigJava;
import org.apache.samza.container.grouper.task.GrouperMetadata;
@@ -34,7 +32,7 @@ import org.apache.samza.coordinator.JobCoordinatorListener;
import org.apache.samza.runtime.LocationId;
import org.apache.samza.runtime.LocationIdProvider;
import org.apache.samza.runtime.LocationIdProviderFactory;
-import org.apache.samza.runtime.ProcessorIdGenerator;
+import org.apache.samza.metrics.MetricsRegistry;
import org.apache.samza.storage.ChangelogStreamManager;
import org.apache.samza.system.StreamMetadataCache;
import org.apache.samza.system.SystemAdmins;
@@ -73,8 +71,8 @@ public class PassthroughJobCoordinator implements JobCoordinator {
private final LocationId locationId;
private JobCoordinatorListener coordinatorListener = null;
- public PassthroughJobCoordinator(Config config) {
- this.processorId = createProcessorId(config);
+ public PassthroughJobCoordinator(String processorId, Config config, MetricsRegistry metricsRegistry) {
+ this.processorId = processorId;
this.config = config;
LocationIdProviderFactory locationIdProviderFactory = Util.getObj(new JobConfig(config).getLocationIdProviderFactory(), LocationIdProviderFactory.class);
LocationIdProvider locationIdProvider = locationIdProviderFactory.getLocationIdProvider(config);
@@ -105,6 +103,7 @@ public class PassthroughJobCoordinator implements JobCoordinator {
coordinatorListener.onNewJobModel(processorId, jobModel);
}
} else {
+ LOGGER.info("JobModel: {} does not contain processorId: {}. Stopping the JobCoordinator", jobModel, processorId);
stop();
}
}
@@ -141,20 +140,4 @@ public class PassthroughJobCoordinator implements JobCoordinator {
public String getProcessorId() {
return this.processorId;
}
-
- private String createProcessorId(Config config) {
- // TODO: This check to be removed after 0.13+
- ApplicationConfig appConfig = new ApplicationConfig(config);
- if (appConfig.getProcessorId() != null) {
- return appConfig.getProcessorId();
- } else if (appConfig.getAppProcessorIdGeneratorClass() != null) {
- ProcessorIdGenerator idGenerator =
- Util.getObj(appConfig.getAppProcessorIdGeneratorClass(), ProcessorIdGenerator.class);
- return idGenerator.generateProcessorId(config);
- } else {
- throw new ConfigException(String
- .format("Expected either %s or %s to be configured", ApplicationConfig.PROCESSOR_ID,
- ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS));
- }
- }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinatorFactory.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinatorFactory.java b/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinatorFactory.java
index 2ba56a6..5d5fecf 100644
--- a/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinatorFactory.java
+++ b/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinatorFactory.java
@@ -21,10 +21,11 @@ package org.apache.samza.standalone;
import org.apache.samza.config.Config;
import org.apache.samza.coordinator.JobCoordinator;
import org.apache.samza.coordinator.JobCoordinatorFactory;
+import org.apache.samza.metrics.MetricsRegistry;
public class PassthroughJobCoordinatorFactory implements JobCoordinatorFactory {
@Override
- public JobCoordinator getJobCoordinator(Config config) {
- return new PassthroughJobCoordinator(config);
+ public JobCoordinator getJobCoordinator(String processorId, Config config, MetricsRegistry metricsRegistry) {
+ return new PassthroughJobCoordinator(processorId, config, metricsRegistry);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java
index 8c5a3ba..8371070 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java
@@ -29,14 +29,10 @@ import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import org.I0Itec.zkclient.IZkStateListener;
-import org.apache.commons.lang3.StringUtils;
import org.apache.samza.checkpoint.CheckpointManager;
-import org.apache.samza.config.ApplicationConfig;
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.MetricsConfig;
import org.apache.samza.config.TaskConfigJava;
import org.apache.samza.config.StorageConfig;
import org.apache.samza.config.ZkConfig;
@@ -52,18 +48,14 @@ import org.apache.samza.job.model.ContainerModel;
import org.apache.samza.job.model.JobModel;
import org.apache.samza.job.model.TaskModel;
import org.apache.samza.metrics.MetricsRegistry;
-import org.apache.samza.metrics.MetricsReporter;
-import org.apache.samza.metrics.ReadableMetricsRegistry;
import org.apache.samza.runtime.LocationId;
import org.apache.samza.runtime.LocationIdProvider;
import org.apache.samza.runtime.LocationIdProviderFactory;
-import org.apache.samza.runtime.ProcessorIdGenerator;
import org.apache.samza.storage.ChangelogStreamManager;
import org.apache.samza.system.StreamMetadataCache;
import org.apache.samza.system.SystemAdmins;
import org.apache.samza.system.SystemStream;
import org.apache.samza.system.SystemStreamPartition;
-import org.apache.samza.util.MetricsReporterLoader;
import org.apache.samza.util.SystemClock;
import org.apache.samza.util.Util;
import org.apache.samza.zk.ZkUtils.ProcessorNode;
@@ -100,7 +92,6 @@ public class ZkJobCoordinator implements JobCoordinator {
private final Config config;
private final ZkBarrierForVersionUpgrade barrier;
private final ZkJobCoordinatorMetrics metrics;
- private final Map<String, MetricsReporter> reporters;
private final ZkLeaderElector leaderElector;
private final AtomicBoolean initiatedShutdown = new AtomicBoolean(false);
private final StreamMetadataCache streamMetadataCache;
@@ -120,11 +111,11 @@ public class ZkJobCoordinator implements JobCoordinator {
@VisibleForTesting
StreamPartitionCountMonitor streamPartitionCountMonitor = null;
- ZkJobCoordinator(Config config, MetricsRegistry metricsRegistry, ZkUtils zkUtils) {
+ ZkJobCoordinator(String processorId, Config config, MetricsRegistry metricsRegistry, ZkUtils zkUtils) {
this.config = config;
this.metrics = new ZkJobCoordinatorMetrics(metricsRegistry);
- this.processorId = createProcessorId(config);
+ this.processorId = processorId;
this.zkUtils = zkUtils;
// setup a listener for a session state change
// we are mostly interested in "session closed" and "new session created" events
@@ -132,7 +123,6 @@ public class ZkJobCoordinator implements JobCoordinator {
leaderElector = new ZkLeaderElector(processorId, zkUtils);
leaderElector.setLeaderElectorListener(new LeaderElectorListenerImpl());
this.debounceTimeMs = new JobConfig(config).getDebounceTimeMs();
- this.reporters = MetricsReporterLoader.getMetricsReporters(new MetricsConfig(config), processorId);
debounceTimer = new ScheduleAfterDebounceTime(processorId);
debounceTimer.setScheduledTaskCallback(throwable -> {
LOG.error("Received exception in debounce timer! Stopping the job coordinator", throwable);
@@ -152,7 +142,6 @@ public class ZkJobCoordinator implements JobCoordinator {
zkUtils.validateZkVersion();
zkUtils.validatePaths(new String[]{keyBuilder.getProcessorsPath(), keyBuilder.getJobModelVersionPath(), keyBuilder.getJobModelPathPrefix(), keyBuilder.getTaskLocalityPath()});
- startMetrics();
systemAdmins.start();
leaderElector.tryBecomeLeader();
zkUtils.subscribeToJobModelVersionChange(new ZkJobModelVersionChangeHandler(zkUtils));
@@ -192,9 +181,6 @@ public class ZkJobCoordinator implements JobCoordinator {
LOG.debug("Shutting down system admins.");
systemAdmins.stop();
- LOG.debug("Shutting down metrics.");
- shutdownMetrics();
-
if (streamPartitionCountMonitor != null) {
streamPartitionCountMonitor.stop();
}
@@ -217,19 +203,6 @@ public class ZkJobCoordinator implements JobCoordinator {
}
}
- private void startMetrics() {
- for (MetricsReporter reporter: reporters.values()) {
- reporter.register("job-coordinator-" + processorId, (ReadableMetricsRegistry) metrics.getMetricsRegistry());
- reporter.start();
- }
- }
-
- private void shutdownMetrics() {
- for (MetricsReporter reporter: reporters.values()) {
- reporter.stop();
- }
- }
-
@Override
public void setListener(JobCoordinatorListener listener) {
this.coordinatorListener = listener;
@@ -307,22 +280,6 @@ public class ZkJobCoordinator implements JobCoordinator {
debounceTimer.scheduleAfterDebounceTime(ON_ZK_CLEANUP, 0, () -> zkUtils.cleanupZK(NUM_VERSIONS_TO_LEAVE));
}
- private String createProcessorId(Config config) {
- // TODO: This check to be removed after 0.13+
- ApplicationConfig appConfig = new ApplicationConfig(config);
- if (appConfig.getProcessorId() != null) {
- return appConfig.getProcessorId();
- } else if (StringUtils.isNotBlank(appConfig.getAppProcessorIdGeneratorClass())) {
- ProcessorIdGenerator idGenerator =
- Util.getObj(appConfig.getAppProcessorIdGeneratorClass(), ProcessorIdGenerator.class);
- return idGenerator.generateProcessorId(config);
- } else {
- throw new ConfigException(String
- .format("Expected either %s or %s to be configured", ApplicationConfig.PROCESSOR_ID,
- ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS));
- }
- }
-
/**
* Generate new JobModel when becoming a leader or the list of processor changed.
*/
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java
index 41294a3..fdc2b0d 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java
@@ -27,7 +27,6 @@ import org.apache.samza.config.ZkConfig;
import org.apache.samza.coordinator.JobCoordinator;
import org.apache.samza.coordinator.JobCoordinatorFactory;
import org.apache.samza.metrics.MetricsRegistry;
-import org.apache.samza.metrics.MetricsRegistryMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,13 +43,12 @@ public class ZkJobCoordinatorFactory implements JobCoordinatorFactory {
* @return An instance of {@link ZkJobCoordinator}
*/
@Override
- public JobCoordinator getJobCoordinator(Config config) {
+ public JobCoordinator getJobCoordinator(String processorId, Config config, MetricsRegistry metricsRegistry) {
// TODO: Separate JC related configs into a "ZkJobCoordinatorConfig"
- MetricsRegistry metricsRegistry = new MetricsRegistryMap();
String jobCoordinatorZkBasePath = getJobCoordinationZkPath(config);
ZkUtils zkUtils = getZkUtils(config, metricsRegistry, jobCoordinatorZkBasePath);
LOG.debug("Creating ZkJobCoordinator with config: {}.", config);
- return new ZkJobCoordinator(config, metricsRegistry, zkUtils);
+ return new ZkJobCoordinator(processorId, config, metricsRegistry, zkUtils);
}
private ZkUtils getZkUtils(Config config, MetricsRegistry metricsRegistry, String coordinatorZkBasePath) {
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-core/src/test/java/org/apache/samza/processor/TestStreamProcessor.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/processor/TestStreamProcessor.java b/samza-core/src/test/java/org/apache/samza/processor/TestStreamProcessor.java
index 1c33b96..d977504 100644
--- a/samza-core/src/test/java/org/apache/samza/processor/TestStreamProcessor.java
+++ b/samza-core/src/test/java/org/apache/samza/processor/TestStreamProcessor.java
@@ -93,7 +93,7 @@ public class TestStreamProcessor {
ProcessorLifecycleListener processorListener,
JobCoordinator jobCoordinator,
SamzaContainer container) {
- super(config, customMetricsReporters, streamTaskFactory, processorListener, jobCoordinator);
+ super("TEST_PROCESSOR_ID", config, customMetricsReporters, streamTaskFactory, processorListener, jobCoordinator);
this.container = container;
}
@@ -325,7 +325,7 @@ public class TestStreamProcessor {
JobCoordinator mockJobCoordinator = Mockito.mock(JobCoordinator.class);
Mockito.doNothing().when(mockJobCoordinator).start();
ProcessorLifecycleListener lifecycleListener = Mockito.mock(ProcessorLifecycleListener.class);
- StreamProcessor streamProcessor = new StreamProcessor(new MapConfig(), new HashMap<>(), null, lifecycleListener, mockJobCoordinator);
+ StreamProcessor streamProcessor = new StreamProcessor("TestProcessorId", new MapConfig(), new HashMap<>(), null, lifecycleListener, mockJobCoordinator);
assertEquals(State.NEW, streamProcessor.getState());
streamProcessor.start();
@@ -344,7 +344,7 @@ public class TestStreamProcessor {
ProcessorLifecycleListener lifecycleListener = Mockito.mock(ProcessorLifecycleListener.class);
SamzaContainer mockSamzaContainer = Mockito.mock(SamzaContainer.class);
MapConfig config = new MapConfig(ImmutableMap.of("task.shutdown.ms", "0"));
- StreamProcessor streamProcessor = new StreamProcessor(config, new HashMap<>(), null, lifecycleListener, mockJobCoordinator);
+ StreamProcessor streamProcessor = new StreamProcessor("TestProcessorId", config, new HashMap<>(), null, lifecycleListener, mockJobCoordinator);
/**
* Without a SamzaContainer running in StreamProcessor and current StreamProcessor state is STARTED,
@@ -412,7 +412,7 @@ public class TestStreamProcessor {
ProcessorLifecycleListener lifecycleListener = Mockito.mock(ProcessorLifecycleListener.class);
SamzaContainer mockSamzaContainer = Mockito.mock(SamzaContainer.class);
MapConfig config = new MapConfig(ImmutableMap.of("task.shutdown.ms", "0"));
- StreamProcessor streamProcessor = PowerMockito.spy(new StreamProcessor(config, new HashMap<>(), null, lifecycleListener, mockJobCoordinator));
+ StreamProcessor streamProcessor = PowerMockito.spy(new StreamProcessor("TestProcessorId", config, new HashMap<>(), null, lifecycleListener, mockJobCoordinator));
Mockito.doNothing().when(mockJobCoordinator).stop();
Mockito.doNothing().when(mockSamzaContainer).shutdown();
@@ -434,7 +434,7 @@ public class TestStreamProcessor {
ProcessorLifecycleListener lifecycleListener = Mockito.mock(ProcessorLifecycleListener.class);
SamzaContainer mockSamzaContainer = Mockito.mock(SamzaContainer.class);
MapConfig config = new MapConfig(ImmutableMap.of("task.shutdown.ms", "0"));
- StreamProcessor streamProcessor = new StreamProcessor(config, new HashMap<>(), null, lifecycleListener, mockJobCoordinator);
+ StreamProcessor streamProcessor = new StreamProcessor("TestProcessorId", config, new HashMap<>(), null, lifecycleListener, mockJobCoordinator);
Exception failureException = new Exception("dummy exception");
@@ -455,7 +455,7 @@ public class TestStreamProcessor {
JobCoordinator mockJobCoordinator = Mockito.mock(JobCoordinator.class);
ProcessorLifecycleListener lifecycleListener = Mockito.mock(ProcessorLifecycleListener.class);
MapConfig config = new MapConfig(ImmutableMap.of("task.shutdown.ms", "0"));
- StreamProcessor streamProcessor = new StreamProcessor(config, new HashMap<>(), null, lifecycleListener, mockJobCoordinator);
+ StreamProcessor streamProcessor = new StreamProcessor("TestProcessorId", config, new HashMap<>(), null, lifecycleListener, mockJobCoordinator);
streamProcessor.state = State.RUNNING;
streamProcessor.jobCoordinatorListener.onCoordinatorStop();
@@ -468,7 +468,7 @@ public class TestStreamProcessor {
public void testStreamProcessorWithStreamProcessorListenerFactory() {
AtomicReference<MockStreamProcessorLifecycleListener> mockListener = new AtomicReference<>();
StreamProcessor streamProcessor =
- new StreamProcessor(mock(Config.class), new HashMap<>(), mock(TaskFactory.class), Optional.empty(),
+ new StreamProcessor("TestProcessorId", mock(Config.class), new HashMap<>(), mock(TaskFactory.class), Optional.empty(),
Optional.empty(), Optional.empty(),
sp -> mockListener.updateAndGet(old -> new MockStreamProcessorLifecycleListener(sp)),
mock(JobCoordinator.class));
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-core/src/test/java/org/apache/samza/runtime/MockProcessorIdGenerator.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/runtime/MockProcessorIdGenerator.java b/samza-core/src/test/java/org/apache/samza/runtime/MockProcessorIdGenerator.java
new file mode 100644
index 0000000..a55a2ab
--- /dev/null
+++ b/samza-core/src/test/java/org/apache/samza/runtime/MockProcessorIdGenerator.java
@@ -0,0 +1,29 @@
+/*
+ * 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.runtime;
+
+import org.apache.samza.config.Config;
+
+public class MockProcessorIdGenerator implements ProcessorIdGenerator {
+ @Override
+ public String generateProcessorId(Config config) {
+ return "testProcessorId";
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java
----------------------------------------------------------------------
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 5bd7893..c691500 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
@@ -19,6 +19,7 @@
package org.apache.samza.runtime;
+import com.google.common.collect.ImmutableMap;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
@@ -32,6 +33,7 @@ import org.apache.samza.application.descriptors.ApplicationDescriptorUtil;
import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.ApplicationConfig;
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.context.ExternalContext;
@@ -42,6 +44,7 @@ import org.apache.samza.task.IdentityStreamTask;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -221,6 +224,29 @@ public class TestLocalApplicationRunner {
assertFalse("Application finished before the timeout.", finished);
}
+ @Test
+ public void testCreateProcessorIdShouldReturnProcessorIdDefinedInConfiguration() {
+ String processorId = "testProcessorId";
+ MapConfig configMap = new MapConfig(ImmutableMap.of(ApplicationConfig.PROCESSOR_ID, processorId));
+ String actualProcessorId = LocalApplicationRunner.createProcessorId(new ApplicationConfig(configMap));
+ assertEquals(processorId, actualProcessorId);
+ }
+
+ @Test
+ public void testCreateProcessorIdShouldInvokeProcessorIdGeneratorDefinedInConfiguration() {
+ String processorId = "testProcessorId";
+ MapConfig configMap = new MapConfig(ImmutableMap.of(ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS, MockProcessorIdGenerator.class.getCanonicalName()));
+ String actualProcessorId = LocalApplicationRunner.createProcessorId(new ApplicationConfig(configMap));
+ assertEquals(processorId, actualProcessorId);
+ }
+
+ @Test(expected = ConfigException.class)
+ public void testCreateProcessorIdShouldThrowExceptionWhenProcessorIdAndGeneratorAreNotDefined() {
+ ApplicationConfig mockConfig = Mockito.mock(ApplicationConfig.class);
+ Mockito.when(mockConfig.getProcessorId()).thenReturn(null);
+ LocalApplicationRunner.createProcessorId(mockConfig);
+ }
+
private void prepareTest() {
ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc =
ApplicationDescriptorUtil.getAppDescriptor(mockApp, config);
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-core/src/test/java/org/apache/samza/zk/TestZkJobCoordinator.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/zk/TestZkJobCoordinator.java b/samza-core/src/test/java/org/apache/samza/zk/TestZkJobCoordinator.java
index 083caad..e0a0941 100644
--- a/samza-core/src/test/java/org/apache/samza/zk/TestZkJobCoordinator.java
+++ b/samza-core/src/test/java/org/apache/samza/zk/TestZkJobCoordinator.java
@@ -53,7 +53,7 @@ public class TestZkJobCoordinator {
when(zkUtils.getZkClient()).thenReturn(mockZkClient);
when(zkUtils.getJobModel(TEST_JOB_MODEL_VERSION)).thenReturn(new JobModel(new MapConfig(), new HashMap<>()));
- ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator(new MapConfig(), new NoOpMetricsRegistry(), zkUtils));
+ ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator("TEST_PROCESSOR_ID", new MapConfig(), new NoOpMetricsRegistry(), zkUtils));
doAnswer(new Answer<Void>() {
public Void answer(InvocationOnMock invocation) {
jcShutdownLatch.countDown();
@@ -80,7 +80,7 @@ public class TestZkJobCoordinator {
ScheduleAfterDebounceTime mockDebounceTimer = Mockito.mock(ScheduleAfterDebounceTime.class);
- ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator(new MapConfig(), new NoOpMetricsRegistry(), zkUtils));
+ ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator("TEST_PROCESSOR_ID", new MapConfig(), new NoOpMetricsRegistry(), zkUtils));
zkJobCoordinator.debounceTimer = mockDebounceTimer;
final ZkSessionStateChangedListener zkSessionStateChangedListener = zkJobCoordinator.new ZkSessionStateChangedListener();
@@ -104,7 +104,7 @@ public class TestZkJobCoordinator {
ScheduleAfterDebounceTime mockDebounceTimer = Mockito.mock(ScheduleAfterDebounceTime.class);
- ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator(new MapConfig(), new NoOpMetricsRegistry(), zkUtils));
+ ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator("TEST_PROCESSOR_ID", new MapConfig(), new NoOpMetricsRegistry(), zkUtils));
StreamPartitionCountMonitor monitor = Mockito.mock(StreamPartitionCountMonitor.class);
zkJobCoordinator.debounceTimer = mockDebounceTimer;
zkJobCoordinator.streamPartitionCountMonitor = monitor;
@@ -127,7 +127,7 @@ public class TestZkJobCoordinator {
ScheduleAfterDebounceTime mockDebounceTimer = Mockito.mock(ScheduleAfterDebounceTime.class);
- ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator(new MapConfig(), new NoOpMetricsRegistry(), zkUtils));
+ ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator("TEST_PROCESSOR_ID", new MapConfig(), new NoOpMetricsRegistry(), zkUtils));
StreamPartitionCountMonitor monitor = Mockito.mock(StreamPartitionCountMonitor.class);
zkJobCoordinator.debounceTimer = mockDebounceTimer;
@@ -154,7 +154,7 @@ public class TestZkJobCoordinator {
ScheduleAfterDebounceTime mockDebounceTimer = Mockito.mock(ScheduleAfterDebounceTime.class);
- ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator(new MapConfig(), new NoOpMetricsRegistry(), zkUtils));
+ ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator("TEST_PROCESSOR_ID", new MapConfig(), new NoOpMetricsRegistry(), zkUtils));
StreamPartitionCountMonitor monitor = Mockito.mock(StreamPartitionCountMonitor.class);
zkJobCoordinator.debounceTimer = mockDebounceTimer;
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessorBase.java
----------------------------------------------------------------------
diff --git a/samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessorBase.java b/samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessorBase.java
index 7bd99bb..3760edb 100644
--- a/samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessorBase.java
+++ b/samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessorBase.java
@@ -47,6 +47,7 @@ import org.apache.samza.config.ZkConfig;
import org.apache.samza.context.Context;
import org.apache.samza.coordinator.JobCoordinator;
import org.apache.samza.coordinator.JobCoordinatorFactory;
+import org.apache.samza.metrics.MetricsRegistryMap;
import org.apache.samza.runtime.ProcessorLifecycleListener;
import org.apache.samza.system.IncomingMessageEnvelope;
import org.apache.samza.system.OutgoingMessageEnvelope;
@@ -132,7 +133,7 @@ public class TestZkStreamProcessorBase extends StandaloneIntegrationTestHarness
map.put(ApplicationConfig.PROCESSOR_ID, pId);
Config config = new MapConfig(map);
String jobCoordinatorFactoryClassName = new JobCoordinatorConfig(config).getJobCoordinatorFactoryClassName();
- JobCoordinator jobCoordinator = Util.getObj(jobCoordinatorFactoryClassName, JobCoordinatorFactory.class).getJobCoordinator(config);
+ JobCoordinator jobCoordinator = Util.getObj(jobCoordinatorFactoryClassName, JobCoordinatorFactory.class).getJobCoordinator(pId, config, new MetricsRegistryMap());
ProcessorLifecycleListener listener = new ProcessorLifecycleListener() {
@Override
@@ -165,7 +166,7 @@ public class TestZkStreamProcessorBase extends StandaloneIntegrationTestHarness
};
StreamProcessor processor =
- new StreamProcessor(config, new HashMap<>(), (StreamTaskFactory) TestStreamTask::new, listener, jobCoordinator);
+ new StreamProcessor(pId, config, new HashMap<>(), (StreamTaskFactory) TestStreamTask::new, listener, jobCoordinator);
return processor;
}
http://git-wip-us.apache.org/repos/asf/samza/blob/df56f2dc/samza-test/src/test/java/org/apache/samza/test/processor/TestStreamProcessor.java
----------------------------------------------------------------------
diff --git a/samza-test/src/test/java/org/apache/samza/test/processor/TestStreamProcessor.java b/samza-test/src/test/java/org/apache/samza/test/processor/TestStreamProcessor.java
index e7040ca..cf41148 100644
--- a/samza-test/src/test/java/org/apache/samza/test/processor/TestStreamProcessor.java
+++ b/samza-test/src/test/java/org/apache/samza/test/processor/TestStreamProcessor.java
@@ -59,6 +59,9 @@ import static org.mockito.Mockito.mock;
public class TestStreamProcessor extends StandaloneIntegrationTestHarness {
+
+ public static final String PROCESSOR_ID = "1";
+
/**
* Testing a basic identity stream task - reads data from a topic and writes it to another topic
* (without any modifications)
@@ -136,7 +139,7 @@ public class TestStreamProcessor extends StandaloneIntegrationTestHarness {
final String outputTopic = "output4";
final int messageCount = 20;
- final Map<String, String> configMap = createConfigs("1", testSystem, inputTopic, outputTopic, messageCount);
+ final Map<String, String> configMap = createConfigs(PROCESSOR_ID, testSystem, inputTopic, outputTopic, messageCount);
configMap.remove("task.class");
final Config configs = new MapConfig(configMap);
final TestStubs stubs = new TestStubs(configs, (StreamTaskFactory) null, bootstrapServers());
@@ -243,12 +246,12 @@ public class TestStreamProcessor extends StandaloneIntegrationTestHarness {
TestStubs(Config config, StreamTaskFactory taskFactory, String bootstrapServer) {
this(bootstrapServer);
- processor = new StreamProcessor(config, new HashMap<>(), taskFactory, listener);
+ processor = new StreamProcessor("1", config, new HashMap<>(), taskFactory, listener);
}
TestStubs(Config config, AsyncStreamTaskFactory taskFactory, String bootstrapServer) {
this(bootstrapServer);
- processor = new StreamProcessor(config, new HashMap<>(), taskFactory, listener);
+ processor = new StreamProcessor("1", config, new HashMap<>(), taskFactory, listener);
}
private void initConsumer(String bootstrapServer) {