You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by na...@apache.org on 2016/12/24 01:03:58 UTC
[1/7] samza git commit: Pulling out from samza-li StandAloneApi
Repository: samza
Updated Branches:
refs/heads/samza-standalone a47e8819f -> 4918e3ad7
Pulling out from samza-li StandAloneApi
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/7d6332b6
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/7d6332b6
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/7d6332b6
Branch: refs/heads/samza-standalone
Commit: 7d6332b69f02b9bdb5fda78983ef64eded79f66b
Parents: a47e881
Author: navina <na...@apache.org>
Authored: Fri Dec 23 16:58:14 2016 -0800
Committer: navina <na...@apache.org>
Committed: Fri Dec 23 16:58:14 2016 -0800
----------------------------------------------------------------------
build.gradle | 1 +
gradle/dependency-versions.gradle | 2 +-
.../org/apache/samza/config/JavaJobConfig.java | 27 +++
.../samza/config/JobCoordinatorConfig.java | 24 ++
.../org/apache/samza/config/TaskConfigJava.java | 14 ++
.../java/org/apache/samza/config/ZkConfig.java | 30 +++
.../samza/coordinator/JobCoordinator.java | 58 +++++
.../coordinator/JobCoordinatorFactory.java | 30 +++
.../leaderelection/LeaderElector.java | 7 +
.../processor/SamzaContainerController.java | 127 ++++++++++
.../apache/samza/processor/StreamProcessor.java | 148 ++++++++++++
.../standalone/StandaloneJobCoordinator.java | 121 ++++++++++
.../StandaloneJobCoordinatorFactory.java | 31 +++
.../samza/zk/BarrierForVersionUpgrade.java | 9 +
.../samza/zk/ScheduleAfterDebounceTime.java | 56 +++++
.../samza/zk/ZkBarrierForVersionUpgrade.java | 147 ++++++++++++
.../java/org/apache/samza/zk/ZkController.java | 11 +
.../org/apache/samza/zk/ZkControllerImpl.java | 136 +++++++++++
.../org/apache/samza/zk/ZkJobCoordinator.java | 198 ++++++++++++++++
.../samza/zk/ZkJobCoordinatorFactory.java | 37 +++
.../java/org/apache/samza/zk/ZkKeyBuilder.java | 45 ++++
.../org/apache/samza/zk/ZkLeaderElector.java | 110 +++++++++
.../java/org/apache/samza/zk/ZkListener.java | 11 +
.../main/java/org/apache/samza/zk/ZkUtils.java | 237 +++++++++++++++++++
.../apache/samza/container/SamzaContainer.scala | 41 +++-
.../samza/job/local/ThreadJobFactory.scala | 11 +-
26 files changed, 1659 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/build.gradle
----------------------------------------------------------------------
diff --git a/build.gradle b/build.gradle
index 5b41c52..0d60970 100644
--- a/build.gradle
+++ b/build.gradle
@@ -159,6 +159,7 @@ project(":samza-core_$scalaVersion") {
compile "net.sf.jopt-simple:jopt-simple:$joptSimpleVersion"
compile "org.codehaus.jackson:jackson-mapper-asl:$jacksonVersion"
compile "org.eclipse.jetty:jetty-webapp:$jettyVersion"
+ compile "com.101tec:zkclient:$zkClientVersion"
testCompile "junit:junit:$junitVersion"
testCompile "org.mockito:mockito-all:$mockitoVersion"
testCompile "org.scalatest:scalatest_$scalaVersion:$scalaTestVersion"
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/gradle/dependency-versions.gradle
----------------------------------------------------------------------
diff --git a/gradle/dependency-versions.gradle b/gradle/dependency-versions.gradle
index 976a49c..872ae1b 100644
--- a/gradle/dependency-versions.gradle
+++ b/gradle/dependency-versions.gradle
@@ -26,7 +26,7 @@
mockitoVersion = "1.8.4"
scalaTestVersion = "2.2.4"
zkClientVersion = "0.8"
- zookeeperVersion = "3.3.4"
+ zookeeperVersion = "3.4.6"
metricsVersion = "2.2.0"
kafkaVersion = "0.10.0.1"
commonsHttpClientVersion = "3.1"
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/config/JavaJobConfig.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/config/JavaJobConfig.java b/samza-core/src/main/java/org/apache/samza/config/JavaJobConfig.java
new file mode 100644
index 0000000..c0747f0
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/config/JavaJobConfig.java
@@ -0,0 +1,27 @@
+package org.apache.samza.config;
+
+public class JavaJobConfig extends MapConfig {
+ private static final String JOB_NAME = "job.name"; // streaming.job_name
+ private static final String JOB_ID = "job.id"; // streaming.job_id
+ private static final String DEFAULT_JOB_ID = "1";
+
+ public JavaJobConfig (Config config) {
+ super(config);
+ }
+
+ public String getJobName() {
+ if (!containsKey(JOB_NAME)) {
+ throw new ConfigException("Missing " + JOB_NAME + " config!");
+ }
+ return get(JOB_NAME);
+ }
+
+ public String getJobName(String defaultValue) {
+ return get(JOB_NAME, defaultValue);
+ }
+
+ public String getJobId() {
+ return get(JOB_ID, DEFAULT_JOB_ID);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/config/JobCoordinatorConfig.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/config/JobCoordinatorConfig.java b/samza-core/src/main/java/org/apache/samza/config/JobCoordinatorConfig.java
new file mode 100644
index 0000000..c8e496e
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/config/JobCoordinatorConfig.java
@@ -0,0 +1,24 @@
+package org.apache.samza.config;
+
+import com.google.common.base.Strings;
+import org.apache.samza.coordinator.JobCoordinatorFactory;
+import org.apache.samza.util.Util;
+
+public class JobCoordinatorConfig extends MapConfig {
+ // TODO: Change this to job-coordinator.factory
+ private static final String JOB_COORDINATOR_FACTORY = "job.coordinator.factory";
+
+ public JobCoordinatorConfig (Config config) {
+ super(config);
+ }
+
+ public String getJobCoordinatorFactoryClassName() {
+ String jobCoordinatorFactoryClassName = get(JOB_COORDINATOR_FACTORY);
+ if (Strings.isNullOrEmpty(jobCoordinatorFactoryClassName)) {
+ throw new ConfigException(
+ String.format("Missing config - %s. Cannot start StreamProcessor!", JOB_COORDINATOR_FACTORY));
+ }
+
+ return jobCoordinatorFactoryClassName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/config/TaskConfigJava.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/config/TaskConfigJava.java b/samza-core/src/main/java/org/apache/samza/config/TaskConfigJava.java
index 648fe58..a88e1ec 100644
--- a/samza-core/src/main/java/org/apache/samza/config/TaskConfigJava.java
+++ b/samza-core/src/main/java/org/apache/samza/config/TaskConfigJava.java
@@ -35,6 +35,10 @@ import scala.collection.JavaConversions;
public class TaskConfigJava extends MapConfig {
+ // Task Configs
+ private static final String TASK_SHUTDOWN_MS = "task.shutdown.ms";
+ public static final long DEFAULT_TASK_SHUTDOWN_MS = 5000L;
+
// broadcast streams consumed by all tasks. e.g. kafka.foo#1
public static final String BROADCAST_INPUT_STREAMS = "task.broadcast.inputs";
private static final String BROADCAST_STREAM_PATTERN = "^[\\d]+$";
@@ -117,4 +121,14 @@ public class TaskConfigJava extends MapConfig {
return Collections.unmodifiableSet(allInputSS);
}
+
+ /**
+ * Returns a value indicating how long to wait for the tasks to shutdown
+ *
+ * @return value indicating how long to wait for the tasks to shutdown
+ */
+ public long getShutdownMs() {
+ if (get(TASK_SHUTDOWN_MS) == null) return DEFAULT_TASK_SHUTDOWN_MS;
+ return Long.valueOf(get(TASK_SHUTDOWN_MS));
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java b/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java
new file mode 100644
index 0000000..973db42
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java
@@ -0,0 +1,30 @@
+package org.apache.samza.config;
+
+public class ZkConfig extends MapConfig {
+ // Connection string for ZK, format: :<hostname>:<port>,..."
+ public static final String ZK_CONNECT = "coordinator.zk.connect";
+ public static final String ZK_SESSION_TIMEOUT_MS = "coordinator.zk.session-timeout-ms";
+ public static final String ZK_CONNECTION_TIMEOUT_MS = "coordinator.zk.session-timeout-ms";
+
+ public static final int DEFAULT_CONNECTION_TIMEOUT_MS = 60000;
+ public static final int DEFAULT_SESSION_TIMEOUT_MS = 30000;
+
+ public ZkConfig (Config config) {
+ super(config);
+ }
+
+ public String getZkConnect() {
+ if (!containsKey(ZK_CONNECT)) {
+ throw new ConfigException("Missing " + ZK_CONNECT + " config!");
+ }
+ return get(ZK_CONNECT);
+ }
+
+ public int getZkSessionTimeoutMs() {
+ return getInt(ZK_SESSION_TIMEOUT_MS, DEFAULT_SESSION_TIMEOUT_MS);
+ }
+
+ public int getZkConnectionTimeoutMs() {
+ return getInt(ZK_CONNECTION_TIMEOUT_MS, DEFAULT_CONNECTION_TIMEOUT_MS);
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/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
new file mode 100644
index 0000000..ce0de2b
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinator.java
@@ -0,0 +1,58 @@
+/*
+ * 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.coordinator;
+
+import org.apache.samza.job.model.JobModel;
+
+/**
+ * A JobCoordinator is a pluggable module in each process that provides the JobModel and the ID to the StreamProcessor.
+ * In some cases, ID assignment is completely config driven, while in other cases, ID assignment may require
+ * coordination with JobCoordinators of other StreamProcessors.
+ * */
+public interface JobCoordinator {
+ /**
+ * Starts the JobCoordinator which involves one or more of the following:
+ * * LeaderElector Module initialization, if any
+ * * If leader, generate JobModel. Else, read JobModel
+ */
+ void start();
+
+ /**
+ * Cleanly shutting down the JobCoordinator involves:
+ * * Shutting down the LeaderElection module (TBD: details depending on leader or not)
+ * * TBD
+ */
+ void stop();
+
+ /**
+ * Returns the logical ID assigned to the processor
+ * This may be specified by the user when used as a Library and hence, it is upto the user to ensure that different
+ * instances of StreamProcessor have unique processor ID. In all other cases, this will be assigned by the leader?? (Need to think more)
+ * @return integer representing the logical processor ID
+ */
+ int getProcessorId();
+
+ /**
+ * Returns the current JobModel
+ * The implementation of the JobCoordinator in the leader needs to know how to read the config and generate JobModel
+ * In case of a non-leader, the JobCoordinator should simply fetch the jobmodel
+ * @return instance of JobModel that describes the partition distribution among the processors (and hence, tasks)
+ */
+ JobModel getJobModel();
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/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
new file mode 100644
index 0000000..af2aaa7
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinatorFactory.java
@@ -0,0 +1,30 @@
+/*
+ * 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.coordinator;
+
+import org.apache.samza.config.Config;
+import org.apache.samza.processor.SamzaContainerController;
+
+public interface JobCoordinatorFactory {
+ /**
+ * @param config Configs relevant for the JobCoordinator TODO: Separate JC related configs into a "JobCoordinatorConfig"
+ * @return An instance of IJobCoordinator
+ */
+ JobCoordinator getJobCoordinator(int processorId, Config config, SamzaContainerController containerController);
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/coordinator/leaderelection/LeaderElector.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/coordinator/leaderelection/LeaderElector.java b/samza-core/src/main/java/org/apache/samza/coordinator/leaderelection/LeaderElector.java
new file mode 100644
index 0000000..fc3cac9
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/coordinator/leaderelection/LeaderElector.java
@@ -0,0 +1,7 @@
+package org.apache.samza.coordinator.leaderelection;
+
+public interface LeaderElector {
+ boolean tryBecomeLeader();
+ void resignLeadership();
+ boolean amILeader();
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java b/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java
new file mode 100644
index 0000000..9352f27
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java
@@ -0,0 +1,127 @@
+package org.apache.samza.processor;
+
+import org.apache.samza.config.ClusterManagerConfig;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.TaskConfigJava;
+import org.apache.samza.container.LocalityManager;
+import org.apache.samza.container.SamzaContainer;
+import org.apache.samza.container.SamzaContainer$;
+import org.apache.samza.job.model.ContainerModel;
+import org.apache.samza.metrics.JmxServer;
+import org.apache.samza.metrics.MetricsReporter;
+import org.apache.samza.util.Util;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class SamzaContainerController {
+ private static final Logger log = LoggerFactory.getLogger(SamzaContainerController.class);
+
+ private final ExecutorService executorService;
+ private volatile SamzaContainer container;
+ private final Map<String, MetricsReporter> metricsReporterMap;
+ private final Object taskFactory;
+ private final long containerShutdownMs;
+
+ // Internal Member Variables
+ private Future containerFuture;
+
+ /**
+ * Creates an instance of a controller for instantiating, starting and/or stopping {@link SamzaContainer}
+ * Requests to execute a container are submitted to the {@link ExecutorService}
+ *
+ * @param taskFactory Factory that be used create instances of {@link org.apache.samza.task.StreamTask} or
+ * {@link org.apache.samza.task.AsyncStreamTask}
+ * @param containerShutdownMs How long the Samza container should wait for an orderly shutdown of task instances
+ * @param metricsReporterMap Map of metric reporter name and {@link MetricsReporter} instance
+ */
+ public SamzaContainerController (
+ Object taskFactory,
+ long containerShutdownMs,
+ Map<String, MetricsReporter> metricsReporterMap) {
+ this.executorService = Executors.newSingleThreadExecutor();
+ this.taskFactory = taskFactory;
+ this.metricsReporterMap = metricsReporterMap;
+ if (containerShutdownMs == -1) {
+ this.containerShutdownMs = TaskConfigJava.DEFAULT_TASK_SHUTDOWN_MS;
+ } else {
+ this.containerShutdownMs = containerShutdownMs;
+ }
+ }
+
+ /**
+ * Instantiates a container and submits to the executor. This method does not actually wait for the container to
+ * fully start-up. For such a behavior, see {@link #awaitStart(long)}
+ *
+ * <b>Note:</b> <i>This method does not stop a currently running container, if any. It is left up to the caller to
+ * ensure that the container has been stopped with stopContainer before invoking this method.</i>
+ *
+ * @param containerModel {@link ContainerModel} instance to use for the current run of the Container
+ * @param config Complete configuration map used by the Samza job
+ * @param maxChangelogStreamPartitions Max number of partitions expected in the changelog streams
+ * TODO: Try to get rid of maxChangelogStreamPartitions from method arguments
+ */
+ public void startContainer(ContainerModel containerModel, Config config, int maxChangelogStreamPartitions) {
+ LocalityManager localityManager = null;
+ if (new ClusterManagerConfig(config).getHostAffinityEnabled()) {
+ localityManager = SamzaContainer$.MODULE$.getLocalityManager(containerModel.getContainerId(), config);
+ }
+ container = SamzaContainer$.MODULE$.apply(
+ containerModel.getContainerId(),
+ containerModel,
+ config,
+ maxChangelogStreamPartitions,
+ localityManager,
+ new JmxServer(),
+ Util.<String, MetricsReporter>javaMapAsScalaMap(metricsReporterMap),
+ taskFactory);
+ containerFuture = executorService.submit(() -> container.run());
+ }
+
+ /**
+ * Method waits for a specified amount of time for the container to fully start-up, which consists of class-loading
+ * all the components and start message processing
+ *
+ * @param timeoutMs Maximum time to wait, in milliseconds
+ * @return {@code true}, if the container started within the specified wait time and {@code false} if the waiting
+ * time elapsed
+ * @throws InterruptedException if the current thread is interrupted while waiting for container to start-up
+ */
+ public boolean awaitStart(long timeoutMs) throws InterruptedException {
+ return container.awaitStart(timeoutMs);
+ }
+
+ /**
+ * Stops a running container, if any. Invoking this method multiple times does not have any side-effects.
+ */
+ public void stopContainer() {
+ if(container == null) {
+ log.warn("Shutdown before a container was created.");
+ return;
+ }
+
+ container.shutdown();
+ try {
+ containerFuture.get(containerShutdownMs, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException | ExecutionException e) {
+ log.error("Ran into problems while trying to stop the container in the processor!", e);
+ } catch (TimeoutException e) {
+ log.warn("Got Timeout Exception while trying to stop the container in the processor! The processor may not shutdown properly", e);
+ }
+ }
+
+ /**
+ * Shutsdown the controller by first stop any running container and then, shutting down the {@link ExecutorService}
+ */
+ public void shutdown() {
+ stopContainer();
+ executorService.shutdown();
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/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
new file mode 100644
index 0000000..0f34400
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java
@@ -0,0 +1,148 @@
+/*
+ * 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.processor;
+
+import org.apache.samza.config.Config;
+import org.apache.samza.config.JobCoordinatorConfig;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.config.TaskConfigJava;
+import org.apache.samza.coordinator.JobCoordinator;
+import org.apache.samza.coordinator.JobCoordinatorFactory;
+import org.apache.samza.metrics.MetricsReporter;
+import org.apache.samza.util.Util;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * StreamProcessor can be embedded in any application or executed in a distributed environment (aka cluster) as
+ * independent processes <br />
+ * <p>
+ * <b>Usage Example:</b>
+ * <pre>
+ * StreamProcessor processor = new StreamProcessor(1, config); <br />
+ * processor.start();
+ * try {
+ * boolean status = processor.awaitStart(TIMEOUT_MS); // Optional - blocking call
+ * if (!status) {
+ * // Timed out
+ * }
+ * ...
+ * } catch (InterruptedException ie) {
+ * ...
+ * } finally {
+ * processor.stop();
+ * }
+ * </pre>
+ */
+public class StreamProcessor {
+ private static final Logger log = LoggerFactory.getLogger(StreamProcessor.class);
+ /**
+ * processor.id is equivalent to containerId in samza. It is a logical identifier used by Samza for a processor.
+ * In a distributed environment, this logical identifier is mapped to a physical identifier of the resource. For
+ * example, Yarn provides a "containerId" for every resource it allocates.
+ * In an embedded environment, this identifier is provided by the user by directly using the StreamProcessor API.
+ * <p>
+ * <b>Note:</b>This identifier has to be unique across the instances of StreamProcessors.
+ */
+ private static final String PROCESSOR_ID = "processor.id";
+ private final int processorId;
+ private final JobCoordinator jobCoordinator;
+ private final SamzaContainerController containerController;
+
+ /**
+ * Create an instance of StreamProcessor that encapsulates a JobCoordinator and Samza Container
+ * <p>
+ * JobCoordinator controls how the various StreamProcessor instances belonging to a job coordinate. It is also
+ * responsible generating and updating JobModel.
+ * When StreamProcessor starts, it starts the JobCoordinator and brings up a SamzaContainer based on the JobModel.
+ * SamzaContainer is executed using an ExecutorService. <br />
+ * <p>
+ * <b>Note:</b> Lifecycle of the ExecutorService is fully managed by the StreamProcessor, and NOT exposed to the user
+ *
+ * @param processorId Unique identifier for a processor within the job. It has the same semantics as
+ * "containerId" in Samza
+ * @param config Instance of config object - contains all configuration required for processing
+ * @param customMetricsReporters Map of custom MetricReporter instances that are to be injected in the Samza job
+ */
+ public StreamProcessor(int processorId, Config config, Map<String, MetricsReporter> customMetricsReporters) {
+ this(processorId, config, customMetricsReporters, (Object) null);
+ }
+
+ private StreamProcessor(int processorId, Config config, Map<String, MetricsReporter> customMetricsReporters,
+ Object taskFactory) {
+ this.processorId = processorId;
+
+ Map<String, String> updatedConfigMap = new HashMap<>();
+ updatedConfigMap.putAll(config);
+ updatedConfigMap.put(PROCESSOR_ID, String.valueOf(processorId));
+ Config updatedConfig = new MapConfig(updatedConfigMap);
+
+
+ this.containerController = new SamzaContainerController(
+ taskFactory,
+ new TaskConfigJava(updatedConfig).getShutdownMs(),
+ customMetricsReporters);
+
+ this.jobCoordinator = Util.
+ <JobCoordinatorFactory>getObj(
+ new JobCoordinatorConfig(updatedConfig)
+ .getJobCoordinatorFactoryClassName())
+ .getJobCoordinator(processorId, updatedConfig, this.containerController);
+ }
+
+ /**
+ * StreamProcessor Lifecycle: start()
+ * <ul>
+ * <li>Starts the JobCoordinator and fetches the JobModel</li>
+ * <li>Starts the container using ContainerModel based on the processorId </li>
+ * </ul>
+ * When start() returns, it only guarantees that the container is initialized and submitted by the controller to
+ * execute
+ */
+ public void start() {
+ jobCoordinator.start();
+ }
+
+ /**
+ * Method that allows the user to wait for a specified amount of time for the container to initialize and start
+ * processing messages
+ *
+ * @param timeoutMs Maximum time to wait, in milliseconds
+ * @return {@code true}, if the container started within the specified wait time and {@code false} if the waiting time
+ * elapsed
+ * @throws InterruptedException if the current thread is interrupted while waiting for container to start-up
+ */
+ public boolean awaitStart(long timeoutMs) throws InterruptedException {
+ return containerController.awaitStart(timeoutMs); // TODO: Should awaitStart be part of the JC interface, instead of directly using container controller
+ }
+
+ /**
+ * StreamProcessor Lifecycle: stop()
+ * <ul>
+ * <li>Stops the SamzaContainer execution</li>
+ * <li>Stops the JobCoordinator</li>
+ * </ul>
+ */
+ public void stop() {
+ jobCoordinator.stop();
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/standalone/StandaloneJobCoordinator.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/standalone/StandaloneJobCoordinator.java b/samza-core/src/main/java/org/apache/samza/standalone/StandaloneJobCoordinator.java
new file mode 100644
index 0000000..7fe1422
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/standalone/StandaloneJobCoordinator.java
@@ -0,0 +1,121 @@
+/*
+ * 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.standalone;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.samza.SamzaException;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.JavaSystemConfig;
+import org.apache.samza.coordinator.JobCoordinator;
+import org.apache.samza.coordinator.JobModelManager;
+import org.apache.samza.coordinator.JobModelManager$;
+import org.apache.samza.job.model.JobModel;
+import org.apache.samza.processor.SamzaContainerController;
+import org.apache.samza.system.StreamMetadataCache;
+import org.apache.samza.system.SystemAdmin;
+import org.apache.samza.system.SystemFactory;
+import org.apache.samza.util.SystemClock;
+import org.apache.samza.util.Util;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Standalone Job Coordinator does not implement any leader elector module or cluster manager
+ *
+ * It generates the JobModel using the Config passed into the constructor.
+ * Since the standalone JobCoordinator does not perform partition management, it allows two kinds of partition
+ * distribution mechanism - consumer-managed partition distribution and user-defined fixed partition distribution.
+ *
+ * */
+public class StandaloneJobCoordinator implements JobCoordinator {
+ private static final Logger log = LoggerFactory.getLogger(StandaloneJobCoordinator.class);
+ private final int processorId;
+ private final Config config;
+ private final JobModelManager jobModelManager;
+ private final SamzaContainerController containerController;
+
+ @VisibleForTesting
+ StandaloneJobCoordinator(
+ int processorId,
+ Config config,
+ SamzaContainerController containerController,
+ JobModelManager jobModelManager) {
+ this.processorId = processorId;
+ this.config = config;
+ this.containerController = containerController;
+ this.jobModelManager = jobModelManager;
+ }
+
+ public StandaloneJobCoordinator(int processorId, Config config, SamzaContainerController containerController) {
+ this.processorId = processorId;
+ this.config = config;
+ this.containerController = containerController;
+
+ JavaSystemConfig systemConfig = new JavaSystemConfig(this.config);
+ Map<String, SystemAdmin> systemAdmins = new HashMap<>();
+ for (String systemName: systemConfig.getSystemNames()) {
+ String systemFactoryClassName = systemConfig.getSystemFactory(systemName);
+ if (systemFactoryClassName == null) {
+ log.error(String.format("A stream uses system %s, which is missing from the configuration.", systemName));
+ throw new SamzaException(String.format("A stream uses system %s, which is missing from the configuration.", systemName));
+ }
+ SystemFactory systemFactory = Util.getObj(systemFactoryClassName);
+ systemAdmins.put(systemName, systemFactory.getAdmin(systemName, this.config));
+ }
+
+ StreamMetadataCache streamMetadataCache = new StreamMetadataCache(Util.<String, SystemAdmin>javaMapAsScalaMap(systemAdmins), 5000, SystemClock.instance());
+
+ /** TODO:
+ * Locality Manager seems to be required in JC for reading locality info and grouping tasks intelligently and also,
+ * in SamzaContainer for writing locality info to the coordinator stream. This closely couples together
+ * TaskNameGrouper with the LocalityManager! Hence, groupers should be a property of the jobcoordinator
+ * (job.coordinator.task.grouper, instead of task.systemstreampartition.grouper)
+ */
+ this.jobModelManager = JobModelManager$.MODULE$.getJobCoordinator(this.config, null, null, streamMetadataCache, null);
+ }
+
+ @Override
+ public void start() {
+ // No-op
+ JobModel jobModel = getJobModel();
+ containerController.startContainer(
+ jobModel.getContainers().get(processorId),
+ jobModel.getConfig(),
+ jobModel.maxChangeLogStreamPartitions);
+ }
+
+ @Override
+ public void stop() {
+ // No-op
+ containerController.shutdown();
+ }
+
+ @Override
+ public int getProcessorId() {
+ return this.processorId;
+ }
+
+ @Override
+ public JobModel getJobModel() {
+ return jobModelManager.jobModel();
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/standalone/StandaloneJobCoordinatorFactory.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/standalone/StandaloneJobCoordinatorFactory.java b/samza-core/src/main/java/org/apache/samza/standalone/StandaloneJobCoordinatorFactory.java
new file mode 100644
index 0000000..7ca85c0
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/standalone/StandaloneJobCoordinatorFactory.java
@@ -0,0 +1,31 @@
+/*
+ * 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.standalone;
+
+import org.apache.samza.config.Config;
+import org.apache.samza.coordinator.JobCoordinator;
+import org.apache.samza.coordinator.JobCoordinatorFactory;
+import org.apache.samza.processor.SamzaContainerController;
+
+public class StandaloneJobCoordinatorFactory implements JobCoordinatorFactory {
+ @Override
+ public JobCoordinator getJobCoordinator(int processorId, Config config, SamzaContainerController containerController) {
+ return new StandaloneJobCoordinator(processorId, config, containerController);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/zk/BarrierForVersionUpgrade.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/BarrierForVersionUpgrade.java b/samza-core/src/main/java/org/apache/samza/zk/BarrierForVersionUpgrade.java
new file mode 100644
index 0000000..691aced
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/zk/BarrierForVersionUpgrade.java
@@ -0,0 +1,9 @@
+package org.apache.samza.zk;
+
+import java.util.List;
+
+
+public interface BarrierForVersionUpgrade {
+ void leaderStartBarrier(String version, List<String> processorsNames);
+ void waitForBarrier(String version, String processorsName, Runnable callback);
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java b/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java
new file mode 100644
index 0000000..1854de6
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java
@@ -0,0 +1,56 @@
+package org.apache.samza.zk;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ScheduleAfterDebounceTime {
+ public static final Logger LOG = LoggerFactory.getLogger(ScheduleAfterDebounceTime.class);
+ public static final long timeoutMs = 1000*10;
+
+ public static final String JOB_MODEL_VERSION_CHANGE = "JobModelVersionChange";
+ public static final String ON_PROCESSOR_CHANGE = "OnProcessorChange";
+ public static final String ON_DATA_CHANGE_ON = "OnDataChanteOn";
+ public static final int DEBOUNCE_TIME_MS = 2000;
+
+
+ private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
+ private final Map<String, ScheduledFuture> futureHandles = new HashMap<>();
+
+ public ScheduleAfterDebounceTime () {
+
+ }
+
+ synchronized public void scheduleAfterDebounceTime (String actionName, long debounceTimeMs, Runnable runnable) {//, final ReadyToCreateJobModelListener listener) {
+ // check if this action has been scheduled already
+ ScheduledFuture sf = futureHandles.get(actionName);
+ if(sf != null && !sf.isDone()) {
+ LOG.info(">>>>>>>>>>>DEBOUNCE: cancel future for " + actionName);
+ // attempt to cancel
+ if(! sf.cancel(false) ) {
+ try {
+ sf.get(timeoutMs, TimeUnit.MILLISECONDS);
+ } catch (Exception e) {
+ // we ignore the exception
+ LOG.warn("cancel for action " + actionName + " failed with ", e);
+ }
+ }
+ futureHandles.remove(actionName);
+ }
+ // schedule a new task
+ sf = scheduledExecutorService.schedule(runnable, debounceTimeMs, TimeUnit.MILLISECONDS);
+ LOG.info(">>>>>>>>>>>DEBOUNCE: scheduled " + actionName + " in " + debounceTimeMs);
+ futureHandles.put(actionName, sf);
+ }
+
+ public void stopScheduler() {
+ // shutdown executor service
+ scheduledExecutorService.shutdown();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/zk/ZkBarrierForVersionUpgrade.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkBarrierForVersionUpgrade.java b/samza-core/src/main/java/org/apache/samza/zk/ZkBarrierForVersionUpgrade.java
new file mode 100644
index 0000000..60a06da
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkBarrierForVersionUpgrade.java
@@ -0,0 +1,147 @@
+package org.apache.samza.zk;
+
+import java.util.List;
+import org.I0Itec.zkclient.IZkChildListener;
+import org.I0Itec.zkclient.IZkDataListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class ZkBarrierForVersionUpgrade implements BarrierForVersionUpgrade {
+ private final ZkUtils zkUtils;
+ private final ZkKeyBuilder keyBuilder;
+ private final static String BARRIER_DONE = "done";
+ private final static Logger LOG = LoggerFactory.getLogger(ZkBarrierForVersionUpgrade.class);
+
+ private final ScheduleAfterDebounceTime debounceTimer;
+
+ final private String barrierPrefix;
+
+ public ZkBarrierForVersionUpgrade( ZkUtils zkUtils, ScheduleAfterDebounceTime debounceTimer) {
+ this.zkUtils = zkUtils;
+ keyBuilder = zkUtils.getKeyBuilder();
+
+ barrierPrefix = keyBuilder.getJobModelVersionBarrierPrefix();
+ this.debounceTimer = debounceTimer;
+ }
+
+ @Override
+ public void leaderStartBarrier(String version, List<String> processorsNames) {
+ String barrierPath = String.format("%s/barrier_%s", barrierPrefix, version);
+ String barrierDonePath = String.format("%s/barrier_done", barrierPath);
+ String barrierProcessors = String.format("%s/barrier_processors", barrierPath);
+ String barrier = String.format("%s/%s/barrier", barrierPrefix, version);
+
+ // TODO - do we need a check if it exists - it needs to be deleted?
+ zkUtils.makeSurePersistentPathsExists(new String[]{barrierPrefix, barrierPath, barrierProcessors, barrierDonePath});
+
+ // callback for when the barrier is reached
+ Runnable callback = new Runnable() {
+ @Override
+ public void run() {
+ LOG.info("Writing BARRIER DONE to " + barrierDonePath);
+ zkUtils.getZkClient().writeData(barrierDonePath, BARRIER_DONE);
+ }
+ };
+ // subscribe for processor's list changes
+ LOG.info("Subscribing for child changes at " + barrierProcessors);
+ zkUtils.getZkClient().subscribeChildChanges(barrierProcessors,
+ new ZkBarrierChangeHandler(callback, processorsNames));
+ }
+
+ @Override
+ public void waitForBarrier(String version, String processorsName, Runnable callback) {
+ // if participant makes this call it means it has already stopped the old container and got the new job model.
+ String barrierPath = String.format("%s/barrier_%s", barrierPrefix, version);
+ String barrierDonePath = String.format("%s/barrier_done", barrierPath);
+ String barrierProcessors = String.format("%s/barrier_processors", barrierPath);
+ String barrierProcessorThis = String.format("%s/%s", barrierProcessors, processorsName);
+
+
+ // update the barrier for this processor
+ LOG.info("Creating a child for barrier at " + barrierProcessorThis);
+ zkUtils.getZkClient().createPersistent(barrierProcessorThis);
+
+ // now subscribe for the barrier
+ zkUtils.getZkClient().subscribeDataChanges(barrierDonePath, new ZkBarrierReachedHandler(barrierDonePath, debounceTimer, callback));
+ }
+
+ /**
+ * listener for the subscription.
+ */
+ class ZkBarrierChangeHandler implements IZkChildListener {
+ Runnable callback;
+ List<String> names;
+
+ public ZkBarrierChangeHandler(Runnable callback, List<String> names) {
+ this.callback = callback;
+ this.names = names;
+ }
+
+ @Override
+ public void handleChildChange(String parentPath, List<String> currentChildren) throws Exception {
+ // Find out the event & Log
+ boolean allIn = true;
+
+ if(currentChildren == null) {
+ LOG.info("Got handleChildChange with null currentChildren");
+ return;
+ }
+ // debug
+ StringBuilder sb = new StringBuilder();
+ for (String child : currentChildren) {
+ sb.append(child).append(",");
+ }
+ LOG.info("list of children in the barrier = " + parentPath + ":" + sb.toString());
+ sb = new StringBuilder();
+ for (String child : names) {
+ sb.append(child).append(",");
+ }
+ LOG.info("list of children to compare against = " + parentPath + ":" + sb.toString());
+
+
+ // check if all the names are in
+ for(String n : names) {
+ if(!currentChildren.contains(n)) {
+ LOG.info("node " + n + " is still not in the list ");
+ allIn = false;
+ break;
+ }
+ }
+ if(allIn) {
+ LOG.info("ALl nodes reached the barrier");
+ callback.run(); // all the names have registered
+ }
+ }
+ }
+
+ class ZkBarrierReachedHandler implements IZkDataListener {
+ private final ScheduleAfterDebounceTime debounceTimer;
+ private final String barrierPathDone;
+ private final Runnable callback;
+ public ZkBarrierReachedHandler(String barrierPathDone, ScheduleAfterDebounceTime debounceTimer, Runnable callback) {
+ this.barrierPathDone = barrierPathDone;
+ this.callback = callback;
+ this.debounceTimer = debounceTimer;
+ }
+
+ @Override
+ public void handleDataChange(String dataPath, Object data)
+ throws Exception {
+ String done = (String) data;
+ LOG.info("got notification about barrier path=" + barrierPathDone + "; done=" + done);
+ if (done.equals(BARRIER_DONE)) {
+ zkUtils.unsubscribeDataChanges(barrierPathDone, this);
+ debounceTimer.scheduleAfterDebounceTime(ScheduleAfterDebounceTime.JOB_MODEL_VERSION_CHANGE, 0, callback);
+ } else {
+ // TODO do we need to resubscribe?
+ }
+ }
+
+ @Override
+ public void handleDataDeleted(String dataPath)
+ throws Exception {
+ LOG.warn("barrier done got deleted at " + dataPath);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/zk/ZkController.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkController.java b/samza-core/src/main/java/org/apache/samza/zk/ZkController.java
new file mode 100644
index 0000000..20e55ab
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkController.java
@@ -0,0 +1,11 @@
+package org.apache.samza.zk;
+
+
+public interface ZkController {
+ void register ();
+ boolean isLeader();
+ void notifyJobModelChange(String version);
+ void stop();
+ void listenToProcessorLiveness();
+ String currentJobModelVersion();
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java b/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java
new file mode 100644
index 0000000..fdd1f02
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java
@@ -0,0 +1,136 @@
+package org.apache.samza.zk;
+
+import org.I0Itec.zkclient.IZkChildListener;
+import org.I0Itec.zkclient.IZkDataListener;
+import org.apache.samza.SamzaException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+
+public class ZkControllerImpl implements ZkController {
+ private static final Logger LOG = LoggerFactory.getLogger(ZkControllerImpl.class);
+
+ private String processorIdStr;
+ private final ZkUtils zkUtils;
+ private final ZkListener zkListener;
+ private final ZkLeaderElector leaderElector;
+ private final ScheduleAfterDebounceTime debounceTimer;
+
+ public ZkControllerImpl (String processorIdStr, ZkUtils zkUtils, ScheduleAfterDebounceTime debounceTimer, ZkListener zkListener) {
+ this.processorIdStr = processorIdStr;
+ this.zkUtils = zkUtils;
+ this.zkListener = zkListener;
+ this.leaderElector = new ZkLeaderElector(this.processorIdStr, this.zkUtils, this.zkListener);
+ this.debounceTimer = debounceTimer;
+
+ init();
+ }
+
+ @Override
+ public void register() {
+
+ // TODO - make a loop here with some number of attempts.
+ // possibly split into two method - becomeLeader() and becomeParticipant()
+ boolean isLeader = leaderElector.tryBecomeLeader();
+ if(isLeader) {
+ listenToProcessorLiveness();
+
+ // zkUtils.subscribeToProcessorChange(zkProcessorChangeListener);
+ debounceTimer.scheduleAfterDebounceTime(ScheduleAfterDebounceTime.ON_PROCESSOR_CHANGE,
+ ScheduleAfterDebounceTime.DEBOUNCE_TIME_MS, () -> zkListener.onBecomeLeader()); // RECONSIDER MAKING THIS SYNC CALL
+
+ }
+
+ // subscribe to JobModel version updates
+ zkUtils.subscribeToJobModelVersionChange(new ZkJobModelVersionChangeHandler(debounceTimer));
+ }
+
+ private void init() {
+ ZkKeyBuilder keyBuilder = zkUtils.getKeyBuilder();
+ zkUtils.makeSurePersistentPathsExists(new String[] {
+ keyBuilder.getProcessorsPath(), keyBuilder.getJobModelVersionPath(), keyBuilder.getJobModelPathPrefix()});
+ }
+
+ @Override
+ public boolean isLeader() {
+ return leaderElector.amILeader();
+ }
+
+ @Override
+ public void notifyJobModelChange(String version) {
+ zkListener.onNewJobModelAvailable(version);
+ }
+
+ @Override
+ public void stop() {
+ if (isLeader()) {
+ leaderElector.resignLeadership();
+ }
+ zkUtils.close();
+ }
+
+ @Override
+ public void listenToProcessorLiveness() {
+ zkUtils.subscribeToProcessorChange(new ZkProcessorChangeHandler(debounceTimer));
+ }
+
+ @Override
+ public String currentJobModelVersion() {
+ return zkUtils.getJobModelVersion();
+ }
+
+ // Only by Leader
+ class ZkProcessorChangeHandler implements IZkChildListener {
+ private final ScheduleAfterDebounceTime debounceTimer;
+ public ZkProcessorChangeHandler(ScheduleAfterDebounceTime debounceTimer) {
+ this.debounceTimer = debounceTimer;
+ }
+ /**
+ * Called when the children of the given path changed.
+ *
+ * @param parentPath The parent path
+ * @param currentChilds The children or null if the root node (parent path) was deleted.
+ * @throws Exception
+ */
+ @Override
+ public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
+ LOG.info(
+ "ZkControllerImpl::ZkProcessorChangeHandler::handleChildChange - Path: " + parentPath + " Current Children: "
+ + currentChilds);
+ debounceTimer.scheduleAfterDebounceTime(ScheduleAfterDebounceTime.ON_PROCESSOR_CHANGE,
+ ScheduleAfterDebounceTime.DEBOUNCE_TIME_MS, () -> zkListener.onProcessorChange(currentChilds));
+ }
+ }
+
+ class ZkJobModelVersionChangeHandler implements IZkDataListener {
+ private final ScheduleAfterDebounceTime debounceTimer;
+ public ZkJobModelVersionChangeHandler(ScheduleAfterDebounceTime debounceTimer) {
+ this.debounceTimer = debounceTimer;
+ }
+ /**
+ * called when job model version gets updated
+ * @param dataPath
+ * @param data
+ * @throws Exception
+ */
+ @Override
+ public void handleDataChange(String dataPath, Object data) throws Exception {
+ LOG.info("pid=" + processorIdStr + ". Got notification on version update change. path=" + dataPath + "; data="
+ + (String) data);
+
+ debounceTimer
+ .scheduleAfterDebounceTime(ScheduleAfterDebounceTime.JOB_MODEL_VERSION_CHANGE, 0, () -> notifyJobModelChange((String) data));
+ }
+ @Override
+ public void handleDataDeleted(String dataPath) throws Exception {
+ throw new SamzaException("version update path has been deleted!.");
+ }
+ }
+
+ public void shutdown() {
+ if(debounceTimer != null)
+ debounceTimer.stopScheduler();
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/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
new file mode 100644
index 0000000..8c36ff2
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java
@@ -0,0 +1,198 @@
+package org.apache.samza.zk;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.samza.SamzaException;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.JavaSystemConfig;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.coordinator.JobCoordinator;
+import org.apache.samza.job.model.ContainerModel;
+import org.apache.samza.job.model.JobModel;
+import org.apache.samza.processor.SamzaContainerController;
+import org.apache.samza.system.StreamMetadataCache;
+import org.apache.samza.system.SystemAdmin;
+import org.apache.samza.system.SystemFactory;
+import org.apache.samza.util.SystemClock;
+import org.apache.samza.util.Util;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ * JobCoordinator for stand alone processor managed via Zookeeper.
+ */
+public class ZkJobCoordinator implements JobCoordinator, ZkListener {
+ private static final Logger log = LoggerFactory.getLogger(ZkJobCoordinator.class);
+
+ private final ZkUtils zkUtils;
+ private final int processorId;
+ private final ZkController zkController;
+ private final SamzaContainerController containerController;
+
+ private final BarrierForVersionUpgrade barrier;
+
+
+ /////////////////////////////////////////
+ private JobModel newJobModel;
+ private String newJobModelVersion; // version published in ZK (by the leader)
+ private Config config;
+ private ZkKeyBuilder keyBuilder;
+ private final ScheduleAfterDebounceTime debounceTimer;
+ //JobModelManager jobModelManager;
+
+ public ZkJobCoordinator(int processorId, Config config, ScheduleAfterDebounceTime debounceTimer, ZkUtils zkUtils, SamzaContainerController containerController) {
+ this.zkUtils = zkUtils;
+ this.keyBuilder = zkUtils.getKeyBuilder();
+ this.debounceTimer = debounceTimer;
+ this.processorId = processorId;
+ this.containerController = containerController;
+ this.zkController = new ZkControllerImpl(String.valueOf(processorId), zkUtils, debounceTimer, this);
+ this.config = config;
+
+
+ barrier = new ZkBarrierForVersionUpgrade(zkUtils, debounceTimer); //should not have any state in it
+
+
+
+ // TEMP for model generation
+ //////////////////////////////// NEEDS TO BE REPLACED //////////////////////////////////////
+ JavaSystemConfig systemConfig = new JavaSystemConfig(this.config);
+ Map<String, SystemAdmin> systemAdmins = new HashMap<>();
+ for (String systemName: systemConfig.getSystemNames()) {
+ String systemFactoryClassName = systemConfig.getSystemFactory(systemName);
+ if (systemFactoryClassName == null) {
+ log.error(String.format("A stream uses system %s, which is missing from the configuration.", systemName));
+ throw new SamzaException(String.format("A stream uses system %s, which is missing from the configuration.", systemName));
+ }
+ SystemFactory systemFactory = Util.getObj(systemFactoryClassName);
+ systemAdmins.put(systemName, systemFactory.getAdmin(systemName, this.config));
+ }
+
+ StreamMetadataCache
+ streamMetadataCache = new StreamMetadataCache(Util.<String, SystemAdmin>javaMapAsScalaMap(systemAdmins), 5000, SystemClock
+ .instance());
+
+ //jobModelManager = //JobModelManager$.MODULE$.getJobCoordinator(this.config, null, null, streamMetadataCache, null);
+
+ ////////////////////////////////////////////////////////////////////////////////////////////
+ }
+
+ @Override
+ public void start() {
+ zkController.register();
+ }
+
+ public void cleanupZk() {
+ zkUtils.deleteRoot();
+ }
+
+ @Override
+ public void stop() {
+ zkController.stop();
+ }
+
+ @Override
+ public int getProcessorId()
+ {
+ return processorId;
+ }
+
+ @Override
+ public JobModel getJobModel() {
+ return newJobModel;
+ }
+
+ //////////////////////////////////////////////// LEADER stuff ///////////////////////////
+ @Override
+ public void onBecomeLeader() {
+ log.info("ZkJobCoordinator::onBecomeLeader - I become the leader!");
+ //zkController.listenToProcessorLiveness();
+ // Reset debounce Timer
+
+ // generate JobProcess
+ generateNewJobModel();
+ }
+
+ private void generateNewJobModel() {
+ // get the current list of processors
+ List<String> currentProcessors = zkUtils.getActiveProcessors();
+
+ // get the current version
+ String currentJMVersion = zkUtils.getJobModelVersion();
+ String nextJMVersion;
+ if(currentJMVersion == null)
+ nextJMVersion = "1";
+ else
+ nextJMVersion = Integer.toString(Integer.valueOf(currentJMVersion) + 1);
+ log.info("pid=" + processorId + "generating new model. Version = " + nextJMVersion);
+
+ Map<String, String> configMap = new HashMap<>();
+ Map<Integer, ContainerModel> containers = new HashMap<>();
+ MapConfig config = new MapConfig(configMap);
+ JobModel jobModel = new JobModel(config, containers);
+
+ log.info("pid=" + processorId + "Generated jobModel: " + jobModel);
+
+ // publish the new version
+ zkUtils.publishNewJobModel(nextJMVersion, jobModel);
+ log.info("pid=" + processorId + "published new JobModel ver=" + nextJMVersion + ";jm=" + jobModel);
+
+ // start the barrier for the job model update
+ barrier.leaderStartBarrier(nextJMVersion, currentProcessors);
+
+ // publish new JobModel version
+ zkUtils.publishNewJobModelVersion(currentJMVersion, nextJMVersion);
+ log.info("pid=" + processorId + "published new JobModel ver=" + nextJMVersion);
+ }
+
+ //////////////////////////////////////////////////////////////////////////////////////////////
+ @Override
+ public void onProcessorChange(List<String> processorIds) {
+ // Reset debounce Timer
+ log.info("ZkJobCoordinator::onProcessorChange - Processors changed! List: " + Arrays.toString(processorIds.toArray()));
+ generateNewJobModel();
+ }
+
+ @Override
+ public void onNewJobModelAvailable(final String version) {
+ newJobModelVersion = version;
+ log.info("pid=" + processorId + "new JobModel available");
+ // stop current work
+ containerController.stopContainer();
+ log.info("pid=" + processorId + "new JobModel available.Container stopped.");
+ // get the new job model
+ newJobModel = zkUtils.getJobModel(version);
+ log.info("pid=" + processorId + "new JobModel available. ver=" + version + "; jm = " + newJobModel);
+
+
+ String currentPath = zkUtils.getEphemeralPath();
+
+ String zkProcessorId = keyBuilder.parseIdFromPath(currentPath);
+
+ // update ZK and wait for all the processors to get this new version
+ barrier.waitForBarrier(version, String.valueOf(zkProcessorId), new Runnable() {
+ @Override
+ public void run() {
+ onNewJobModelConfirmed(version);
+ }
+ });
+ }
+
+ @Override
+ public void onNewJobModelConfirmed(String version) {
+ log.info("pid=" + processorId + "new version " + version + " of the job model got confirmed");
+ // get the new Model
+ // ?????
+ JobModel jobModel = getJobModel();
+ log.info("pid=" + processorId + "got the new job model =" + jobModel);
+ /*
+ containerController.startContainer(
+ jobModel.getContainers().get(processorId),
+ jobModel.getConfig(),
+ jobModel.maxChangeLogStreamPartitions);
+ */
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/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
new file mode 100644
index 0000000..90b0097
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinatorFactory.java
@@ -0,0 +1,37 @@
+package org.apache.samza.zk;
+
+import org.apache.samza.config.Config;
+import org.apache.samza.config.JavaJobConfig;
+import org.apache.samza.config.ZkConfig;
+import org.apache.samza.coordinator.JobCoordinator;
+import org.apache.samza.coordinator.JobCoordinatorFactory;
+import org.apache.samza.processor.SamzaContainerController;
+
+public class ZkJobCoordinatorFactory implements JobCoordinatorFactory {
+ /**
+ * Method to instantiate an implementation of JobCoordinator
+ *
+ * @param processorId Indicates the StreamProcessor's id to which this Job Coordinator is associated with
+ * @param config Configs relevant for the JobCoordinator TODO: Separate JC related configs into a "JobCoordinatorConfig"
+ * @return An instance of IJobCoordinator
+ */
+ @Override
+ public JobCoordinator getJobCoordinator(int processorId, Config config, SamzaContainerController containerController) {
+ JavaJobConfig jobConfig = new JavaJobConfig(config);
+ String groupName = String.format("%s-%s", jobConfig.getJobName(), jobConfig.getJobId());
+ ZkConfig zkConfig = new ZkConfig(config);
+ ScheduleAfterDebounceTime debounceTimer = new ScheduleAfterDebounceTime();
+ return new ZkJobCoordinator(
+ processorId,
+ config,
+ debounceTimer,
+ new ZkUtils(
+ new ZkKeyBuilder(groupName),
+ zkConfig.getZkConnect(),
+ debounceTimer,
+ String.valueOf(processorId),
+ zkConfig.getZkSessionTimeoutMs(),
+ zkConfig.getZkConnectionTimeoutMs()),
+ containerController);
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java b/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java
new file mode 100644
index 0000000..7ad62be
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java
@@ -0,0 +1,45 @@
+package org.apache.samza.zk;
+
+public class ZkKeyBuilder {
+ private final String pathPrefix;
+ public static final String PROCESSORS_PATH = "processors";
+
+ public static final String JOBMODEL_VERSION_PATH = "jobModelVersion";
+
+ public ZkKeyBuilder () {
+ this("");
+ }
+ public ZkKeyBuilder (String pathPrefix) {
+ this.pathPrefix = pathPrefix;
+ }
+
+ public String getProcessorsPath() {
+ return String.format("/%s/%s", pathPrefix, PROCESSORS_PATH);
+ }
+
+ public static String parseIdFromPath(String path) {
+ if (path != null)
+ return path.substring(path.indexOf("processor-"));
+ return null;
+ }
+
+ public String getJobModelVersionPath() {
+ return String.format("/%s/%s", pathPrefix, JOBMODEL_VERSION_PATH);
+ }
+
+ public String getJobModelPathPrefix() {
+ return String.format("/%s/jobModels", pathPrefix);
+ }
+
+ public String getJobModelPath(String jobModelVersion) {
+ return String.format("%s/%s", getJobModelPathPrefix(), jobModelVersion);
+ }
+
+ public String getJobModelVersionBarrierPrefix() {
+ return String.format("/%s/versionBarriers", pathPrefix);
+ }
+
+ public String getRootPath() {
+ return "/" + pathPrefix;
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/zk/ZkLeaderElector.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkLeaderElector.java b/samza-core/src/main/java/org/apache/samza/zk/ZkLeaderElector.java
new file mode 100644
index 0000000..8bffeb6
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkLeaderElector.java
@@ -0,0 +1,110 @@
+package org.apache.samza.zk;
+
+import java.util.Arrays;
+import org.I0Itec.zkclient.IZkDataListener;
+import org.apache.samza.SamzaException;
+import org.apache.samza.coordinator.leaderelection.LeaderElector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Random;
+
+
+public class ZkLeaderElector implements LeaderElector {
+ public static final Logger log = LoggerFactory.getLogger(ZkLeaderElector.class);
+ private final ZkUtils zkUtils;
+ private final ZkListener zkListener;
+ private final String processorIdStr;
+ private final ZkKeyBuilder keyBuilder;
+
+ private String leaderId = null;
+ private final ZkLeaderListener zkLeaderListener = new ZkLeaderListener();
+ private String currentSubscription = null;
+ private final Random random = new Random();
+
+ public ZkLeaderElector (String processorIdStr, ZkUtils zkUtils, ZkListener zkListener) {
+ this.processorIdStr = processorIdStr;
+ this.zkUtils = zkUtils;
+ this.keyBuilder = this.zkUtils.getKeyBuilder();
+ this.zkListener = zkListener;
+ }
+
+ @Override
+ public boolean tryBecomeLeader() {
+ String currentPath = zkUtils.getEphemeralPath();
+
+ if (currentPath == null || currentPath.isEmpty()) {
+ zkUtils.registerProcessorAndGetId();
+ currentPath = zkUtils.getEphemeralPath();
+ }
+
+ List<String> children = zkUtils.getActiveProcessors();
+ int index = children.indexOf(ZkKeyBuilder.parseIdFromPath(currentPath));
+
+ if (index == -1) {
+ // Retry register here??
+ throw new SamzaException("Looks like we are no longer connected to Zk. Need to reconnect??");
+ }
+
+ if (index == 0) {
+ log.info("pid=" + processorIdStr + " Eligible to be the leader!");
+ leaderId = ZkKeyBuilder.parseIdFromPath(currentPath);
+ return true;
+ }
+
+ log.info("pid=" + processorIdStr + ";index=" + index + ";children=" + Arrays.toString(children.toArray()) + " Not eligible to be a leader yet!");
+ leaderId = ZkKeyBuilder.parseIdFromPath(children.get(0));
+ String prevCandidate = children.get(index - 1);
+ if (!prevCandidate.equals(currentSubscription)) {
+ if (currentSubscription != null) {
+ zkUtils.unsubscribeDataChanges(keyBuilder.getProcessorsPath() + "/" + currentSubscription, zkLeaderListener);
+ }
+ currentSubscription = prevCandidate;
+ log.info("pid=" + processorIdStr + "Subscribing to " + prevCandidate);
+ zkUtils.subscribeDataChanges(keyBuilder.getProcessorsPath() + "/" + currentSubscription, zkLeaderListener);
+ }
+
+ // Double check that the previous candidate still exists
+ boolean prevCandidateExists = zkUtils.exists(keyBuilder.getProcessorsPath() + "/" + currentSubscription);
+ if (prevCandidateExists) {
+ log.info("pid=" + processorIdStr + "Previous candidate still exists. Continuing as non-leader");
+ } else {
+ // TODO - what actually happens here..
+ try {
+ Thread.sleep(random.nextInt(1000));
+ } catch (InterruptedException e) {
+ Thread.interrupted();
+ }
+ log.info("pid=" + processorIdStr + "Previous candidate doesn't exist anymore. Trying to become leader again...");
+ return tryBecomeLeader();
+ }
+ return false;
+ }
+
+ @Override
+ public void resignLeadership() {
+ }
+
+ @Override
+ public boolean amILeader() {
+ return zkUtils.getEphemeralPath() != null
+ && leaderId != null
+ && leaderId.equals(ZkKeyBuilder.parseIdFromPath(zkUtils.getEphemeralPath()));
+ }
+
+ // Only by non-leaders
+ class ZkLeaderListener implements IZkDataListener {
+
+ @Override
+ public void handleDataChange(String dataPath, Object data) throws Exception {
+ log.info("ZkLeaderListener::handleDataChange on path " + dataPath + " Data: " + data);
+ }
+
+ @Override
+ public void handleDataDeleted(String dataPath) throws Exception {
+ log.info("ZkLeaderListener::handleDataDeleted on path " + dataPath);
+ tryBecomeLeader();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/zk/ZkListener.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkListener.java b/samza-core/src/main/java/org/apache/samza/zk/ZkListener.java
new file mode 100644
index 0000000..4a1c491
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkListener.java
@@ -0,0 +1,11 @@
+package org.apache.samza.zk;
+
+import java.util.List;
+
+public interface ZkListener {
+ void onBecomeLeader();
+ void onProcessorChange(List<String> processorIds);
+
+ void onNewJobModelAvailable(String version); // start job model update (stop current work)
+ void onNewJobModelConfirmed(String version); // start new work according to the new model
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java b/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
new file mode 100644
index 0000000..6655468
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
@@ -0,0 +1,237 @@
+package org.apache.samza.zk;
+
+import java.io.IOException;
+import org.I0Itec.zkclient.IZkChildListener;
+import org.I0Itec.zkclient.IZkDataListener;
+import org.I0Itec.zkclient.IZkStateListener;
+import org.I0Itec.zkclient.ZkClient;
+import org.I0Itec.zkclient.ZkConnection;
+import org.I0Itec.zkclient.exception.ZkInterruptedException;
+import org.apache.samza.SamzaException;
+import org.apache.samza.config.ZkConfig;
+import org.apache.samza.job.model.JobModel;
+import org.apache.samza.serializers.model.SamzaObjectMapper;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class ZkUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(ZkUtils.class);
+
+ public final ReentrantLock lock = new ReentrantLock();
+
+ private final ZkStateChangeHandler zkStateChangeHandler;
+ private final ZkClient zkClient;
+ private final ZkConnection zkConnnection;
+ private volatile String ephemeralPath = null;
+ private final ZkKeyBuilder keyBuilder;
+ private final int sessionTimeoutMs;
+ private final int connectionTimeoutMs;
+ private final ScheduleAfterDebounceTime debounceTimer;
+ private final String processorId;
+
+ public ZkUtils(String zkConnectString, ScheduleAfterDebounceTime debounceTimer, String processorId) {
+ this(new ZkKeyBuilder(), zkConnectString, debounceTimer, processorId, ZkConfig.DEFAULT_SESSION_TIMEOUT_MS, ZkConfig.DEFAULT_CONNECTION_TIMEOUT_MS);
+ }
+
+ public ZkUtils(ZkKeyBuilder zkKeyBuilder, String zkConnectString, ScheduleAfterDebounceTime debounceTimer, String processorId, int sessionTimeoutMs, int connectionTimeoutMs) {
+ this.keyBuilder = zkKeyBuilder;
+ this.sessionTimeoutMs = sessionTimeoutMs;
+ this.connectionTimeoutMs = connectionTimeoutMs;
+ this.zkConnnection = new ZkConnection(zkConnectString, this.sessionTimeoutMs);
+ this.zkClient = new ZkClient(zkConnnection, this.connectionTimeoutMs);
+ this.zkClient.waitForKeeperState(Watcher.Event.KeeperState.SyncConnected, 10000, TimeUnit.MILLISECONDS);
+ this.debounceTimer = debounceTimer;
+ this.zkStateChangeHandler = new ZkStateChangeHandler(debounceTimer);
+ this.processorId = processorId;
+ }
+
+ public void connect() throws ZkInterruptedException {
+ boolean isConnected = zkClient.waitUntilConnected(connectionTimeoutMs, TimeUnit.MILLISECONDS);
+ if (!isConnected) {
+ throw new RuntimeException("Unable to connect to Zookeeper within connectionTimeout " + connectionTimeoutMs + "ms. Shutting down!");
+ } else {
+ zkClient.subscribeStateChanges(zkStateChangeHandler);
+ }
+ }
+
+ public ZkClient getZkClient() {
+ return zkClient;
+ }
+
+ public ZkConnection getZkConnnection() {
+ return zkConnnection;
+ }
+
+ public ZkKeyBuilder getKeyBuilder() {
+ return keyBuilder;
+ }
+ public void makeSurePersistentPathsExists(String[] paths) {
+ for(String path: paths) {
+ if (!zkClient.exists(path)) {
+ zkClient.createPersistent(path, true);
+ }
+ }
+ }
+
+ public synchronized String registerProcessorAndGetId() {
+ try {
+ // TODO: Data should be more than just the hostname. Use Json serialized data
+ ephemeralPath =
+ zkClient.createEphemeralSequential(keyBuilder.getProcessorsPath() + "/processor-", InetAddress.getLocalHost().getHostName());
+ return ephemeralPath;
+ } catch (UnknownHostException e) {
+ throw new RuntimeException("Failed to register as worker. Aborting...");
+ }
+ }
+
+ public synchronized String getEphemeralPath() {
+ return ephemeralPath;
+ }
+
+ public List<String> getActiveProcessors() {
+ List<String> children = zkClient.getChildren(keyBuilder.getProcessorsPath());
+ assert children.size() > 0;
+ Collections.sort(children);
+ LOG.info("Found these children - " + children);
+ return children;
+ }
+
+ ////////////////////////// TEMP ///////////////// NEEDS to be discussed ////////////////
+ public void publishNewJobModel(String jobModelVersion, JobModel jobModel) {
+ try {
+ // We assume (needs to be verified) that this call will FAIL if the node already exists!!!!!!!!
+ ObjectMapper mmapper = SamzaObjectMapper.getObjectMapper();
+ String jobModelStr = mmapper.writeValueAsString(jobModel);
+ LOG.info("pid=" + processorId + " jobModelAsString=" + jobModelStr);
+ zkClient.createPersistent(keyBuilder.getJobModelPath(jobModelVersion), jobModelStr);
+ LOG.info("wrote jobModel path =" + keyBuilder.getJobModelPath(jobModelVersion));
+ } catch (Exception e) {
+ throw new SamzaException(e);
+ }
+ }
+ public JobModel getJobModel(String jobModelVersion) {
+ LOG.info("pid=" + processorId + "read the model ver=" + jobModelVersion + " from " + keyBuilder.getJobModelPath(jobModelVersion));
+ Object data = zkClient.readData(keyBuilder.getJobModelPath(jobModelVersion));
+ ObjectMapper mmapper = SamzaObjectMapper.getObjectMapper();
+ JobModel jm;
+ try {
+ jm = mmapper.readValue((String)data, JobModel.class);
+ } catch (IOException e) {
+ throw new SamzaException("failed to read JobModel from ZK", e);
+ }
+ return jm;
+ }
+ ///////////////////////////////////////////////////////////////////////////
+
+
+ public String getJobModelVersion() {
+ return zkClient.<String>readData(keyBuilder.getJobModelVersionPath());
+ }
+
+ public void publishNewJobModelVersion(String oldVersion, String newVersion) {
+ Stat stat = new Stat();
+ String currentVersion = zkClient.<String>readData(keyBuilder.getJobModelVersionPath(), stat);
+ LOG.info("pid=" + processorId + " publishing new version: " + newVersion + "; oldVersion = " + oldVersion + "(" + stat.getVersion() + ")");
+ if(currentVersion != null && !currentVersion.equals(oldVersion)) {
+ throw new SamzaException("Someone change JMVersion while Leader was generating: expected" + oldVersion + ", got " + currentVersion);
+ }
+ int dataVersion = stat.getVersion();
+ stat = zkClient.writeDataReturnStat(keyBuilder.getJobModelVersionPath(), newVersion, dataVersion);
+ if(stat.getVersion() != dataVersion + 1)
+ throw new SamzaException("Someone changed data version of the JMVersion while Leader was generating a new one. current= " + dataVersion + ", old version = " + stat.getVersion());
+
+ LOG.info("pid=" + processorId +
+ " published new version: " + newVersion + "; expected dataVersion = " + dataVersion + "(" + stat.getVersion()
+ + ")");
+ }
+
+ /**
+ * subscribe for changes of JobModel version
+ * @param dataListener
+ */
+ public void subscribeToJobModelVersionChange(IZkDataListener dataListener) {
+ LOG.info("pid=" + processorId + " subscribing for jm version change at:" + keyBuilder.getJobModelVersionPath());
+ zkClient.subscribeDataChanges(keyBuilder.getJobModelVersionPath(), dataListener);
+ }
+
+ public void subscribeToProcessorChange(IZkChildListener listener) {
+ LOG.info("pid=" + processorId + " subscribing for child change at:" + keyBuilder.getProcessorsPath());
+ zkClient.subscribeChildChanges(keyBuilder.getProcessorsPath(), listener);
+ }
+
+ /* Wrapper for standard I0Itec methods */
+ public void unsubscribeDataChanges(String path, IZkDataListener dataListener) {
+ LOG.info("pid=" + processorId + " unsubscribing for data change at:" + path);
+ zkClient.unsubscribeDataChanges(path, dataListener);
+ }
+
+ public void subscribeDataChanges(String path, IZkDataListener dataListener) {
+ LOG.info("pid=" + processorId + " subscribing for data change at:" + path);
+ zkClient.subscribeDataChanges(path, dataListener);
+ }
+
+ public boolean exists(String path) {
+ return zkClient.exists(path);
+ }
+
+ public void close() {
+ zkClient.close();
+ }
+
+ public void deleteRoot() {
+ String rootPath = keyBuilder.getRootPath();
+ if(rootPath != null && !rootPath.isEmpty() && zkClient.exists(rootPath)) {
+ LOG.info("pid=" + processorId + " Deleteing root: " + rootPath);
+ zkClient.deleteRecursive(rootPath);
+ }
+ }
+
+ class ZkStateChangeHandler implements IZkStateListener {
+ private final ScheduleAfterDebounceTime debounceTimer;
+ public ZkStateChangeHandler(ScheduleAfterDebounceTime debounceTimer) {
+ this.debounceTimer = debounceTimer;
+ }
+
+ /**
+ * Called when the zookeeper connection state has changed.
+ *
+ * @param state The new state.
+ * @throws Exception On any error.
+ */
+ @Override
+ public void handleStateChanged(Watcher.Event.KeeperState state) throws Exception { }
+
+ /**
+ * Called after the zookeeper session has expired and a new session has been created. You would have to re-create
+ * any ephemeral nodes here.
+ *
+ * @throws Exception On any error.
+ */
+ @Override
+ public void handleNewSession() throws Exception {
+
+ }
+
+ /**
+ * Called when a session cannot be re-established. This should be used to implement connection
+ * failure handling e.g. retry to connect or pass the error up
+ *
+ * @param error The error that prevents a session from being established
+ * @throws Exception On any error.
+ */
+ @Override
+ public void handleSessionEstablishmentError(Throwable error) throws Exception {
+
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
index f1d62c5..f4d605f 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
@@ -32,7 +32,7 @@ import org.apache.samza.checkpoint.{CheckpointListener, CheckpointManagerFactory
import org.apache.samza.config.JobConfig.Config2Job
import org.apache.samza.config.MetricsConfig.Config2Metrics
import org.apache.samza.config.SerializerConfig.Config2Serializer
-import org.apache.samza.config.ShellCommandConfig
+import org.apache.samza.config.{Config, ShellCommandConfig}
import org.apache.samza.config.StorageConfig.Config2Storage
import org.apache.samza.config.StreamConfig.Config2Stream
import org.apache.samza.config.SystemConfig.Config2System
@@ -112,7 +112,14 @@ object SamzaContainer extends Logging {
try {
jmxServer = newJmxServer()
- SamzaContainer(containerModel, jobModel, jmxServer).run
+ val containerModel = jobModel.getContainers.get(containerId.toInt)
+ SamzaContainer(
+ containerId.toInt,
+ containerModel,
+ config,
+ jobModel.maxChangeLogStreamPartitions,
+ getLocalityManager(containerId, config),
+ jmxServer).run
} finally {
if (jmxServer != null) {
jmxServer.stop
@@ -120,6 +127,17 @@ object SamzaContainer extends Logging {
}
}
+ def getLocalityManager(containerId: Int, config: Config): LocalityManager = {
+ val containerName = getSamzaContainerName(containerId)
+ val registryMap = new MetricsRegistryMap(containerName)
+ val coordinatorSystemProducer =
+ new CoordinatorStreamSystemFactory()
+ .getCoordinatorStreamSystemProducer(
+ config,
+ new SamzaContainerMetrics(containerName, registryMap).registry)
+ new LocalityManager(coordinatorSystemProducer)
+ }
+
/**
* Fetches config, task:SSP assignments, and task:changelog partition
* assignments, and returns objects to be used for SamzaContainer's
@@ -136,10 +154,19 @@ object SamzaContainer extends Logging {
classOf[JobModel])
}
- def apply(containerModel: ContainerModel, jobModel: JobModel, jmxServer: JmxServer) = {
- val config = jobModel.getConfig
- val containerId = containerModel.getContainerId
- val containerName = "samza-container-%s" format containerId
+ def getSamzaContainerName(containerId: Int): String = {
+ "samza-container-%d" format containerId
+ }
+
+ def apply(
+ containerId: Int,
+ containerModel: ContainerModel,
+ config: Config,
+ maxChangeLogStreamPartitions: Int,
+ localityManager: LocalityManager,
+ jmxServer: JmxServer,
+ customReporters: Map[String, MetricsReporter] = Map[String, MetricsReporter]()) = {
+ val containerName = getSamzaContainerName(containerId)
val containerPID = Util.getContainerPID
info("Setting up Samza container: %s" format containerName)
@@ -528,7 +555,7 @@ object SamzaContainer extends Logging {
taskStores = taskStores,
storeConsumers = storeConsumers,
changeLogSystemStreams = changeLogSystemStreams,
- jobModel.maxChangeLogStreamPartitions,
+ maxChangeLogStreamPartitions,
streamMetadataCache = streamMetadataCache,
storeBaseDir = defaultStoreBaseDir,
loggedStoreBaseDir = loggedStorageBaseDir,
http://git-wip-us.apache.org/repos/asf/samza/blob/7d6332b6/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala b/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala
index eaab3a6..9ccf6fc 100644
--- a/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala
+++ b/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala
@@ -38,7 +38,8 @@ class ThreadJobFactory extends StreamJobFactory with Logging {
def getJob(config: Config): StreamJob = {
info("Creating a ThreadJob, which is only meant for debugging.")
val coordinator = JobModelManager(config)
- val containerModel = coordinator.jobModel.getContainers.get(0)
+ val jobModel = coordinator.jobModel
+ val containerModel = jobModel.getContainers.get(0)
// Give developers a nice friendly warning if they've specified task.opts and are using a threaded job.
config.getTaskOpts match {
@@ -52,7 +53,13 @@ class ThreadJobFactory extends StreamJobFactory with Logging {
override def run(): Unit = {
val jmxServer = new JmxServer
try {
- SamzaContainer(containerModel, coordinator.jobModel, jmxServer).run()
+ SamzaContainer(
+ containerModel.getContainerId,
+ containerModel,
+ config,
+ jobModel.maxChangeLogStreamPartitions,
+ null,
+ new JmxServer)
} finally {
jmxServer.stop
}
[5/7] samza git commit: added ContainerIds list to the group() method
of a groupper
Posted by na...@apache.org.
added ContainerIds list to the group() method of a groupper
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/e37f9105
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/e37f9105
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/e37f9105
Branch: refs/heads/samza-standalone
Commit: e37f91053e7b84a08e250b6ee3b4fd9348a2c624
Parents: 8cbf014
Author: navina <na...@apache.org>
Authored: Fri Dec 23 17:01:52 2016 -0800
Committer: navina <na...@apache.org>
Committed: Fri Dec 23 17:01:52 2016 -0800
----------------------------------------------------------------------
.../task/SimpleGroupByContainerCount.java | 27 ++++++++++++--------
.../container/grouper/task/TaskNameGrouper.java | 5 ++++
.../standalone/StandaloneJobCoordinator.java | 2 +-
.../org/apache/samza/zk/ZkJobCoordinator.java | 23 ++++++++++++-----
.../java/org/apache/samza/zk/ZkKeyBuilder.java | 13 +++++++++-
.../samza/coordinator/JobCoordinator.scala | 18 ++++++++-----
6 files changed, 63 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/e37f9105/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java b/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java
index c9489f7..359c4ed 100644
--- a/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java
+++ b/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java
@@ -13,18 +13,30 @@ import org.apache.samza.job.model.ContainerModel;
import org.apache.samza.job.model.TaskModel;
-public class SimpleGroupByContainerCount implements BalancingTaskNameGrouper {
- private final int containerCount;
+public class SimpleGroupByContainerCount implements TaskNameGrouper {
+ private final int startContainerCount;
public SimpleGroupByContainerCount() {
- this.containerCount = 1;
+ this.startContainerCount = 1;
}
public SimpleGroupByContainerCount(int containerCount) {
if (containerCount <= 0) throw new IllegalArgumentException("Must have at least one container");
- this.containerCount = containerCount;
+ this.startContainerCount = containerCount;
}
@Override
public Set<ContainerModel> group(Set<TaskModel> tasks) {
+ List<Integer> containerIds = new ArrayList<>(startContainerCount);
+ for (int i = 0; i < startContainerCount; i++) {
+ containerIds.add(i);
+ }
+ return group(tasks, containerIds);
+ }
+
+ public Set<ContainerModel> group(Set<TaskModel> tasks, List<Integer> containersIds) {
+ if(containersIds == null)
+ return this.group(tasks);
+
+ int containerCount = containersIds.size();
// Sort tasks by taskName.
List<TaskModel> sortedTasks = new ArrayList<>(tasks);
@@ -43,14 +55,9 @@ public class SimpleGroupByContainerCount implements BalancingTaskNameGrouper {
// Convert to a Set of ContainerModel
Set<ContainerModel> containerModels = new HashSet<>();
for (int i = 0; i < containerCount; i++) {
- containerModels.add(new ContainerModel(i, taskGroups[i]));
+ containerModels.add(new ContainerModel(containersIds.get(i), taskGroups[i]));
}
return Collections.unmodifiableSet(containerModels);
}
-
- @Override
- public Set<ContainerModel> balance(Set<TaskModel> tasks, LocalityManager localityManager) {
- return null;
- }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/e37f9105/samza-core/src/main/java/org/apache/samza/container/grouper/task/TaskNameGrouper.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/container/grouper/task/TaskNameGrouper.java b/samza-core/src/main/java/org/apache/samza/container/grouper/task/TaskNameGrouper.java
index 59a3237..848beea 100644
--- a/samza-core/src/main/java/org/apache/samza/container/grouper/task/TaskNameGrouper.java
+++ b/samza-core/src/main/java/org/apache/samza/container/grouper/task/TaskNameGrouper.java
@@ -18,6 +18,7 @@
*/
package org.apache.samza.container.grouper.task;
+import java.util.List;
import org.apache.samza.job.model.ContainerModel;
import org.apache.samza.job.model.TaskModel;
@@ -50,4 +51,8 @@ public interface TaskNameGrouper {
* @return Set of containers, which contain the tasks that were passed in.
*/
Set<ContainerModel> group(Set<TaskModel> tasks);
+
+ default Set<ContainerModel> group(Set<TaskModel> tasks, List<Integer> containersIds) {
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/e37f9105/samza-core/src/main/java/org/apache/samza/standalone/StandaloneJobCoordinator.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/standalone/StandaloneJobCoordinator.java b/samza-core/src/main/java/org/apache/samza/standalone/StandaloneJobCoordinator.java
index 7fe1422..5c525d1 100644
--- a/samza-core/src/main/java/org/apache/samza/standalone/StandaloneJobCoordinator.java
+++ b/samza-core/src/main/java/org/apache/samza/standalone/StandaloneJobCoordinator.java
@@ -90,7 +90,7 @@ public class StandaloneJobCoordinator implements JobCoordinator {
* TaskNameGrouper with the LocalityManager! Hence, groupers should be a property of the jobcoordinator
* (job.coordinator.task.grouper, instead of task.systemstreampartition.grouper)
*/
- this.jobModelManager = JobModelManager$.MODULE$.getJobCoordinator(this.config, null, null, streamMetadataCache, null);
+ this.jobModelManager = JobModelManager$.MODULE$.getJobCoordinator(this.config, null, null, streamMetadataCache, null, null);
}
@Override
http://git-wip-us.apache.org/repos/asf/samza/blob/e37f9105/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 3014c1b..560e19b 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
@@ -1,5 +1,6 @@
package org.apache.samza.zk;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
@@ -43,6 +44,7 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
private Config config;
private ZkKeyBuilder keyBuilder;
private final ScheduleAfterDebounceTime debounceTimer;
+ private final StreamMetadataCache streamMetadataCache;
JobModelManager jobModelManager;
public ZkJobCoordinator(int processorId, Config config, ScheduleAfterDebounceTime debounceTimer, ZkUtils zkUtils, SamzaContainerController containerController) {
@@ -73,11 +75,10 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
systemAdmins.put(systemName, systemFactory.getAdmin(systemName, this.config));
}
- StreamMetadataCache
- streamMetadataCache = new StreamMetadataCache(Util.<String, SystemAdmin>javaMapAsScalaMap(systemAdmins), 5000, SystemClock
+ streamMetadataCache = new StreamMetadataCache(Util.<String, SystemAdmin>javaMapAsScalaMap(systemAdmins), 5000, SystemClock
.instance());
- jobModelManager = JobModelManager$.MODULE$.getJobCoordinator(this.config, null, null, streamMetadataCache, null);
+
////////////////////////////////////////////////////////////////////////////////////////////
}
@@ -131,10 +132,20 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
nextJMVersion = Integer.toString(Integer.valueOf(currentJMVersion) + 1);
log.info("pid=" + processorId + "generating new model. Version = " + nextJMVersion);
- Map<String, String> configMap = new HashMap<>();
- Map<Integer, ContainerModel> containers = new HashMap<>();
- MapConfig config = new MapConfig(configMap);
+ //Map<String, String> configMap = new HashMap<>();
+ //Map<Integer, ContainerModel> containers = new HashMap<>();
+ //MapConfig config = new MapConfig(configMap);
//JobModel jobModel = new JobModel(config, containers);
+ StringBuilder sb = new StringBuilder();
+ List<Integer> containerIds = new ArrayList<>();
+ for(String processor: currentProcessors){
+ String zkProcessorId = keyBuilder.parseContainerIdFromProcessorId(processor);
+ sb.append(zkProcessorId).append(",");
+ containerIds.add(Integer.valueOf(zkProcessorId));
+ }
+ log.info("processorsIds: " + sb.toString());
+
+ jobModelManager = JobModelManager$.MODULE$.getJobCoordinator(this.config, null, null, streamMetadataCache, null, containerIds);
JobModel jobModel = jobModelManager.jobModel();
log.info("pid=" + processorId + "Generated jobModel: " + jobModel);
http://git-wip-us.apache.org/repos/asf/samza/blob/e37f9105/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java b/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java
index 7ad62be..42d0c86 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java
@@ -1,8 +1,12 @@
package org.apache.samza.zk;
+import org.apache.samza.SamzaException;
+
+
public class ZkKeyBuilder {
private final String pathPrefix;
public static final String PROCESSORS_PATH = "processors";
+ public static final String PROCESSOR_ID_PREFIX = "processor-";
public static final String JOBMODEL_VERSION_PATH = "jobModelVersion";
@@ -19,10 +23,17 @@ public class ZkKeyBuilder {
public static String parseIdFromPath(String path) {
if (path != null)
- return path.substring(path.indexOf("processor-"));
+ return path.substring(path.indexOf(PROCESSOR_ID_PREFIX));
return null;
}
+ public static String parseContainerIdFromProcessorId(String prId) {
+ if(prId == null)
+ throw new SamzaException("processor id is null");
+
+ return prId.substring(prId.indexOf(PROCESSOR_ID_PREFIX) + PROCESSOR_ID_PREFIX.length());
+ }
+
public String getJobModelVersionPath() {
return String.format("/%s/%s", pathPrefix, JOBMODEL_VERSION_PATH);
}
http://git-wip-us.apache.org/repos/asf/samza/blob/e37f9105/samza-core/src/main/scala/org/apache/samza/coordinator/JobCoordinator.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/coordinator/JobCoordinator.scala b/samza-core/src/main/scala/org/apache/samza/coordinator/JobCoordinator.scala
index df63b97..7bf4921 100644
--- a/samza-core/src/main/scala/org/apache/samza/coordinator/JobCoordinator.scala
+++ b/samza-core/src/main/scala/org/apache/samza/coordinator/JobCoordinator.scala
@@ -106,7 +106,7 @@ object JobModelManager extends Logging {
}
}
- val jobCoordinator = getJobCoordinator(config, changelogManager, localityManager, streamMetadataCache, streamPartitionCountMonitor)
+ val jobCoordinator = getJobCoordinator(config, changelogManager, localityManager, streamMetadataCache, streamPartitionCountMonitor, null)
createChangeLogStreams(config, jobCoordinator.jobModel.maxChangeLogStreamPartitions)
jobCoordinator
@@ -121,8 +121,9 @@ object JobModelManager extends Logging {
changelogManager: ChangelogPartitionManager,
localityManager: LocalityManager,
streamMetadataCache: StreamMetadataCache,
- streamPartitionCountMonitor: StreamPartitionCountMonitor) = {
- val jobModel: JobModel = initializeJobModel(config, changelogManager, localityManager, streamMetadataCache)
+ streamPartitionCountMonitor: StreamPartitionCountMonitor,
+ containerIds: java.util.List[Integer]) = {
+ val jobModel: JobModel = initializeJobModel(config, changelogManager, localityManager, streamMetadataCache, containerIds)
jobModelRef.set(jobModel)
val server = new HttpServer
@@ -188,7 +189,8 @@ object JobModelManager extends Logging {
private def initializeJobModel(config: Config,
changelogManager: ChangelogPartitionManager,
localityManager: LocalityManager,
- streamMetadataCache: StreamMetadataCache): JobModel = {
+ streamMetadataCache: StreamMetadataCache,
+ containerIds: java.util.List[Integer]): JobModel = {
// Do grouping to fetch TaskName to SSP mapping
val allSystemStreamPartitions = getMatchedInputStreamPartitions(config, streamMetadataCache)
val grouper = getSystemStreamPartitionGrouper(config)
@@ -215,7 +217,8 @@ object JobModelManager extends Logging {
allSystemStreamPartitions,
groups,
previousChangelogMapping,
- localityManager)
+ localityManager,
+ containerIds)
val jobModel = jobModelGenerator()
@@ -248,7 +251,8 @@ object JobModelManager extends Logging {
allSystemStreamPartitions: util.Set[SystemStreamPartition],
groups: util.Map[TaskName, util.Set[SystemStreamPartition]],
previousChangelogMapping: util.Map[TaskName, Integer],
- localityManager: LocalityManager): JobModel = {
+ localityManager: LocalityManager,
+ containerIds: java.util.List[Integer]): JobModel = {
// If no mappings are present(first time the job is running) we return -1, this will allow 0 to be the first change
// mapping.
@@ -280,7 +284,7 @@ object JobModelManager extends Logging {
if (containerGrouper.isInstanceOf[BalancingTaskNameGrouper])
containerGrouper.asInstanceOf[BalancingTaskNameGrouper].balance(taskModels, localityManager)
else
- containerGrouper.group(taskModels)
+ containerGrouper.group(taskModels, containerIds)
}
val containerMap = asScalaSet(containerModels).map { case (containerModel) => Integer.valueOf(containerModel.getContainerId) -> containerModel }.toMap
[2/7] samza git commit: Merging TaskFactory changes
Posted by na...@apache.org.
Merging TaskFactory changes
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/be6be8cb
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/be6be8cb
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/be6be8cb
Branch: refs/heads/samza-standalone
Commit: be6be8cba1a28a8395ca9c6700eb125093f6c517
Parents: 7d6332b
Author: navina <na...@apache.org>
Authored: Fri Dec 23 16:59:07 2016 -0800
Committer: navina <na...@apache.org>
Committed: Fri Dec 23 16:59:07 2016 -0800
----------------------------------------------------------------------
.../samza/task/AsyncStreamTaskFactory.java | 28 ++++
.../apache/samza/task/StreamTaskFactory.java | 28 ++++
.../apache/samza/container/RunLoopFactory.java | 20 ++-
.../coordinator/JobCoordinatorFactory.java | 3 +
.../apache/samza/processor/StreamProcessor.java | 27 +++-
.../org/apache/samza/task/AsyncRunLoop.java | 37 ++---
.../main/java/org/apache/samza/zk/ZkUtils.java | 2 +-
.../org/apache/samza/container/RunLoop.scala | 18 +--
.../apache/samza/container/SamzaContainer.scala | 137 ++++++++++++-------
.../apache/samza/container/TaskInstance.scala | 21 +--
.../samza/job/local/ThreadJobFactory.scala | 12 +-
.../org/apache/samza/task/TestAsyncRunLoop.java | 20 +--
.../apache/samza/container/TestRunLoop.scala | 15 +-
.../samza/container/TestSamzaContainer.scala | 4 +-
.../samza/container/TestTaskInstance.scala | 6 +-
15 files changed, 251 insertions(+), 127 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/samza-api/src/main/java/org/apache/samza/task/AsyncStreamTaskFactory.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/task/AsyncStreamTaskFactory.java b/samza-api/src/main/java/org/apache/samza/task/AsyncStreamTaskFactory.java
new file mode 100644
index 0000000..e5ce9c4
--- /dev/null
+++ b/samza-api/src/main/java/org/apache/samza/task/AsyncStreamTaskFactory.java
@@ -0,0 +1,28 @@
+/*
+ * 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.task;
+
+/**
+ * Build {@link AsyncStreamTask} instances.
+ * Implementations should return a new instance for each {@link #createInstance()} invocation.
+ */
+public interface AsyncStreamTaskFactory {
+ AsyncStreamTask createInstance();
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/samza-api/src/main/java/org/apache/samza/task/StreamTaskFactory.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/task/StreamTaskFactory.java b/samza-api/src/main/java/org/apache/samza/task/StreamTaskFactory.java
new file mode 100644
index 0000000..ec53bc0
--- /dev/null
+++ b/samza-api/src/main/java/org/apache/samza/task/StreamTaskFactory.java
@@ -0,0 +1,28 @@
+/*
+ * 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.task;
+
+/**
+ * Build {@link StreamTask} instances.
+ * Implementations should return a new instance for each {@link #createInstance()} invocation.
+ */
+public interface StreamTaskFactory {
+ StreamTask createInstance();
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/samza-core/src/main/java/org/apache/samza/container/RunLoopFactory.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/container/RunLoopFactory.java b/samza-core/src/main/java/org/apache/samza/container/RunLoopFactory.java
index 23a68cb..32ab47a 100644
--- a/samza-core/src/main/java/org/apache/samza/container/RunLoopFactory.java
+++ b/samza-core/src/main/java/org/apache/samza/container/RunLoopFactory.java
@@ -19,20 +19,19 @@
package org.apache.samza.container;
-import java.util.concurrent.ExecutorService;
import org.apache.samza.SamzaException;
import org.apache.samza.config.TaskConfig;
-import org.apache.samza.util.HighResolutionClock;
import org.apache.samza.system.SystemConsumers;
import org.apache.samza.task.AsyncRunLoop;
-import org.apache.samza.task.AsyncStreamTask;
-import org.apache.samza.task.StreamTask;
+import org.apache.samza.util.HighResolutionClock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.collection.JavaConversions;
import scala.runtime.AbstractFunction0;
import scala.runtime.AbstractFunction1;
+import java.util.concurrent.ExecutorService;
+
import static org.apache.samza.util.Util.asScalaClock;
/**
@@ -46,7 +45,7 @@ public class RunLoopFactory {
private static final long DEFAULT_COMMIT_MS = 60000L;
private static final long DEFAULT_CALLBACK_TIMEOUT_MS = -1L;
- public static Runnable createRunLoop(scala.collection.immutable.Map<TaskName, TaskInstance<?>> taskInstances,
+ public static Runnable createRunLoop(scala.collection.immutable.Map<TaskName, TaskInstance> taskInstances,
SystemConsumers consumerMultiplexer,
ExecutorService threadPool,
long maxThrottlingDelayMs,
@@ -62,9 +61,9 @@ public class RunLoopFactory {
log.info("Got commit milliseconds: " + taskCommitMs);
- int asyncTaskCount = taskInstances.values().count(new AbstractFunction1<TaskInstance<?>, Object>() {
+ int asyncTaskCount = taskInstances.values().count(new AbstractFunction1<TaskInstance, Object>() {
@Override
- public Boolean apply(TaskInstance<?> t) {
+ public Boolean apply(TaskInstance t) {
return t.isAsyncTask();
}
});
@@ -77,9 +76,8 @@ public class RunLoopFactory {
if (asyncTaskCount == 0) {
log.info("Run loop in single thread mode.");
- scala.collection.immutable.Map<TaskName, TaskInstance<StreamTask>> streamTaskInstances = (scala.collection.immutable.Map) taskInstances;
return new RunLoop(
- streamTaskInstances,
+ taskInstances,
consumerMultiplexer,
containerMetrics,
maxThrottlingDelayMs,
@@ -95,12 +93,10 @@ public class RunLoopFactory {
log.info("Got callback timeout: " + callbackTimeout);
- scala.collection.immutable.Map<TaskName, TaskInstance<AsyncStreamTask>> asyncStreamTaskInstances = (scala.collection.immutable.Map) taskInstances;
-
log.info("Run loop in asynchronous mode.");
return new AsyncRunLoop(
- JavaConversions.mapAsJavaMap(asyncStreamTaskInstances),
+ JavaConversions.mapAsJavaMap(taskInstances),
threadPool,
consumerMultiplexer,
taskMaxConcurrency,
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/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 af2aaa7..e12e49f 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
@@ -23,7 +23,10 @@ import org.apache.samza.processor.SamzaContainerController;
public interface JobCoordinatorFactory {
/**
+ * @param processorId Unique identifier for the processor
* @param config Configs relevant for the JobCoordinator TODO: Separate JC related configs into a "JobCoordinatorConfig"
+ * @param containerController Controller interface for starting and stopping container. In future, it may simply
+ * pause the container and add/remove tasks
* @return An instance of IJobCoordinator
*/
JobCoordinator getJobCoordinator(int processorId, Config config, SamzaContainerController containerController);
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/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 0f34400..61795e1 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
@@ -25,6 +25,8 @@ import org.apache.samza.config.TaskConfigJava;
import org.apache.samza.coordinator.JobCoordinator;
import org.apache.samza.coordinator.JobCoordinatorFactory;
import org.apache.samza.metrics.MetricsReporter;
+import org.apache.samza.task.AsyncStreamTaskFactory;
+import org.apache.samza.task.StreamTaskFactory;
import org.apache.samza.util.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,11 +36,11 @@ import java.util.Map;
/**
* StreamProcessor can be embedded in any application or executed in a distributed environment (aka cluster) as
- * independent processes <br />
+ * independent processes
* <p>
* <b>Usage Example:</b>
* <pre>
- * StreamProcessor processor = new StreamProcessor(1, config); <br />
+ * StreamProcessor processor = new StreamProcessor(1, config);
* processor.start();
* try {
* boolean status = processor.awaitStart(TIMEOUT_MS); // Optional - blocking call
@@ -74,7 +76,7 @@ public class StreamProcessor {
* JobCoordinator controls how the various StreamProcessor instances belonging to a job coordinate. It is also
* responsible generating and updating JobModel.
* When StreamProcessor starts, it starts the JobCoordinator and brings up a SamzaContainer based on the JobModel.
- * SamzaContainer is executed using an ExecutorService. <br />
+ * SamzaContainer is executed using an ExecutorService.
* <p>
* <b>Note:</b> Lifecycle of the ExecutorService is fully managed by the StreamProcessor, and NOT exposed to the user
*
@@ -82,6 +84,25 @@ public class StreamProcessor {
* "containerId" in Samza
* @param config Instance of config object - contains all configuration required for processing
* @param customMetricsReporters Map of custom MetricReporter instances that are to be injected in the Samza job
+ * @param asyncStreamTaskFactory The {@link AsyncStreamTaskFactory} to be used for creating task instances.
+ */
+ public StreamProcessor(int processorId, Config config, Map<String, MetricsReporter> customMetricsReporters,
+ AsyncStreamTaskFactory asyncStreamTaskFactory) {
+ this(processorId, config, customMetricsReporters, (Object) asyncStreamTaskFactory);
+ }
+
+ /**
+ * Same as {@link #StreamProcessor(int, Config, Map, AsyncStreamTaskFactory)}, except task instances are created
+ * using the provided {@link StreamTaskFactory}.
+ */
+ public StreamProcessor(int processorId, Config config, Map<String, MetricsReporter> customMetricsReporters,
+ StreamTaskFactory streamTaskFactory) {
+ this(processorId, config, customMetricsReporters, (Object) streamTaskFactory);
+ }
+
+ /**
+ * Same as {@link #StreamProcessor(int, Config, Map, AsyncStreamTaskFactory)}, except task instances are created
+ * using the "task.class" configuration instead of a task factory.
*/
public StreamProcessor(int processorId, Config config, Map<String, MetricsReporter> customMetricsReporters) {
this(processorId, config, customMetricsReporters, (Object) null);
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/samza-core/src/main/java/org/apache/samza/task/AsyncRunLoop.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/task/AsyncRunLoop.java b/samza-core/src/main/java/org/apache/samza/task/AsyncRunLoop.java
index 16f1563..ab5514c 100644
--- a/samza-core/src/main/java/org/apache/samza/task/AsyncRunLoop.java
+++ b/samza-core/src/main/java/org/apache/samza/task/AsyncRunLoop.java
@@ -75,7 +75,7 @@ public class AsyncRunLoop implements Runnable, Throttleable {
private volatile Throwable throwable = null;
private final HighResolutionClock clock;
- public AsyncRunLoop(Map<TaskName, TaskInstance<AsyncStreamTask>> taskInstances,
+ public AsyncRunLoop(Map<TaskName, TaskInstance> taskInstances,
ExecutorService threadPool,
SystemConsumers consumerMultiplexer,
int maxConcurrency,
@@ -100,7 +100,7 @@ public class AsyncRunLoop implements Runnable, Throttleable {
this.workerTimer = Executors.newSingleThreadScheduledExecutor();
this.clock = clock;
Map<TaskName, AsyncTaskWorker> workers = new HashMap<>();
- for (TaskInstance<AsyncStreamTask> task : taskInstances.values()) {
+ for (TaskInstance task : taskInstances.values()) {
workers.put(task.taskName(), new AsyncTaskWorker(task));
}
// Partions and tasks assigned to the container will not change during the run loop life time
@@ -112,14 +112,12 @@ public class AsyncRunLoop implements Runnable, Throttleable {
* Returns mapping of the SystemStreamPartition to the AsyncTaskWorkers to efficiently route the envelopes
*/
private static Map<SystemStreamPartition, List<AsyncTaskWorker>> getSspToAsyncTaskWorkerMap(
- Map<TaskName, TaskInstance<AsyncStreamTask>> taskInstances, Map<TaskName, AsyncTaskWorker> taskWorkers) {
+ Map<TaskName, TaskInstance> taskInstances, Map<TaskName, AsyncTaskWorker> taskWorkers) {
Map<SystemStreamPartition, List<AsyncTaskWorker>> sspToWorkerMap = new HashMap<>();
- for (TaskInstance<AsyncStreamTask> task : taskInstances.values()) {
+ for (TaskInstance task : taskInstances.values()) {
Set<SystemStreamPartition> ssps = JavaConversions.setAsJavaSet(task.systemStreamPartitions());
for (SystemStreamPartition ssp : ssps) {
- if (sspToWorkerMap.get(ssp) == null) {
- sspToWorkerMap.put(ssp, new ArrayList<AsyncTaskWorker>());
- }
+ sspToWorkerMap.putIfAbsent(ssp, new ArrayList<>());
sspToWorkerMap.get(ssp).add(taskWorkers.get(task.taskName()));
}
}
@@ -202,7 +200,8 @@ public class AsyncRunLoop implements Runnable, Throttleable {
private IncomingMessageEnvelope chooseEnvelope() {
IncomingMessageEnvelope envelope = consumerMultiplexer.choose(false);
if (envelope != null) {
- log.trace("Choose envelope ssp {} offset {} for processing", envelope.getSystemStreamPartition(), envelope.getOffset());
+ log.trace("Choose envelope ssp {} offset {} for processing",
+ envelope.getSystemStreamPartition(), envelope.getOffset());
containerMetrics.envelopes().inc();
} else {
log.trace("No envelope is available");
@@ -310,12 +309,12 @@ public class AsyncRunLoop implements Runnable, Throttleable {
* will run the task asynchronously. It runs window and commit in the provided thread pool.
*/
private class AsyncTaskWorker implements TaskCallbackListener {
- private final TaskInstance<AsyncStreamTask> task;
+ private final TaskInstance task;
private final TaskCallbackManager callbackManager;
private volatile AsyncTaskState state;
- AsyncTaskWorker(TaskInstance<AsyncStreamTask> task) {
+ AsyncTaskWorker(TaskInstance task) {
this.task = task;
this.callbackManager = new TaskCallbackManager(this, callbackTimer, callbackTimeoutMs, maxConcurrency, clock);
Set<SystemStreamPartition> sspSet = getWorkingSSPSet(task);
@@ -352,12 +351,14 @@ public class AsyncRunLoop implements Runnable, Throttleable {
* @param task
* @return a Set of SSPs such that all SSPs are not at end of stream.
*/
- private Set<SystemStreamPartition> getWorkingSSPSet(TaskInstance<AsyncStreamTask> task) {
+ private Set<SystemStreamPartition> getWorkingSSPSet(TaskInstance task) {
Set<SystemStreamPartition> allPartitions = new HashSet<>(JavaConversions.setAsJavaSet(task.systemStreamPartitions()));
// filter only those SSPs that are not at end of stream.
- Set<SystemStreamPartition> workingSSPSet = allPartitions.stream().filter(ssp -> !consumerMultiplexer.isEndOfStream(ssp)).collect(Collectors.toSet());
+ Set<SystemStreamPartition> workingSSPSet = allPartitions.stream()
+ .filter(ssp -> !consumerMultiplexer.isEndOfStream(ssp))
+ .collect(Collectors.toSet());
return workingSSPSet;
}
@@ -512,15 +513,18 @@ public class AsyncRunLoop implements Runnable, Throttleable {
state.doneProcess();
TaskCallbackImpl callbackImpl = (TaskCallbackImpl) callback;
containerMetrics.processNs().update(clock.nanoTime() - callbackImpl.timeCreatedNs);
- log.trace("Got callback complete for task {}, ssp {}", callbackImpl.taskName, callbackImpl.envelope.getSystemStreamPartition());
+ log.trace("Got callback complete for task {}, ssp {}",
+ callbackImpl.taskName, callbackImpl.envelope.getSystemStreamPartition());
TaskCallbackImpl callbackToUpdate = callbackManager.updateCallback(callbackImpl);
if (callbackToUpdate != null) {
IncomingMessageEnvelope envelope = callbackToUpdate.envelope;
- log.trace("Update offset for ssp {}, offset {}", envelope.getSystemStreamPartition(), envelope.getOffset());
+ log.trace("Update offset for ssp {}, offset {}",
+ envelope.getSystemStreamPartition(), envelope.getOffset());
// update offset
- task.offsetManager().update(task.taskName(), envelope.getSystemStreamPartition(), envelope.getOffset());
+ task.offsetManager().update(task.taskName(),
+ envelope.getSystemStreamPartition(), envelope.getOffset());
// update coordinator
coordinatorRequests.update(callbackToUpdate.coordinator);
@@ -697,7 +701,8 @@ public class AsyncRunLoop implements Runnable, Throttleable {
PendingEnvelope pendingEnvelope = pendingEnvelopeQueue.remove();
int queueSize = pendingEnvelopeQueue.size();
taskMetrics.pendingMessages().set(queueSize);
- log.trace("fetch envelope ssp {} offset {} to process.", pendingEnvelope.envelope.getSystemStreamPartition(), pendingEnvelope.envelope.getOffset());
+ log.trace("fetch envelope ssp {} offset {} to process.",
+ pendingEnvelope.envelope.getSystemStreamPartition(), pendingEnvelope.envelope.getOffset());
log.debug("Task {} pending envelopes count is {} after fetching.", taskName, queueSize);
if (pendingEnvelope.markProcessed()) {
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java b/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
index 6655468..0be2b04 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
@@ -157,7 +157,7 @@ public class ZkUtils {
/**
* subscribe for changes of JobModel version
- * @param dataListener
+ * @param dataListener describe this
*/
public void subscribeToJobModelVersionChange(IZkDataListener dataListener) {
LOG.info("pid=" + processorId + " subscribing for jm version change at:" + keyBuilder.getJobModelVersionPath());
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/samza-core/src/main/scala/org/apache/samza/container/RunLoop.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/container/RunLoop.scala b/samza-core/src/main/scala/org/apache/samza/container/RunLoop.scala
index 7df7d88..b1ab1e0 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/RunLoop.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/RunLoop.scala
@@ -22,7 +22,6 @@ package org.apache.samza.container
import org.apache.samza.task.CoordinatorRequests
import org.apache.samza.system.{IncomingMessageEnvelope, SystemConsumers, SystemStreamPartition}
import org.apache.samza.task.ReadableCoordinator
-import org.apache.samza.task.StreamTask
import org.apache.samza.util.{Logging, Throttleable, ThrottlingExecutor, TimerUtils}
import scala.collection.JavaConversions._
@@ -37,7 +36,7 @@ import scala.collection.JavaConversions._
* be done when.
*/
class RunLoop (
- val taskInstances: Map[TaskName, TaskInstance[StreamTask]],
+ val taskInstances: Map[TaskName, TaskInstance],
val consumerMultiplexer: SystemConsumers,
val metrics: SamzaContainerMetrics,
val maxThrottlingDelayMs: Long,
@@ -57,13 +56,16 @@ class RunLoop (
// Keep a mapping of SystemStreamPartition to TaskInstance to efficiently route them.
val systemStreamPartitionToTaskInstances = getSystemStreamPartitionToTaskInstancesMapping
- def getSystemStreamPartitionToTaskInstancesMapping: Map[SystemStreamPartition, List[TaskInstance[StreamTask]]] = {
- // We could just pass in the SystemStreamPartitionMap during construction, but it's safer and cleaner to derive the information directly
- def getSystemStreamPartitionToTaskInstance(taskInstance: TaskInstance[StreamTask]) = taskInstance.systemStreamPartitions.map(_ -> taskInstance).toMap
+ def getSystemStreamPartitionToTaskInstancesMapping: Map[SystemStreamPartition, List[TaskInstance]] = {
+ // We could just pass in the SystemStreamPartitionMap during construction,
+ // but it's safer and cleaner to derive the information directly
+ def getSystemStreamPartitionToTaskInstance(taskInstance: TaskInstance) =
+ taskInstance.systemStreamPartitions.map(_ -> taskInstance).toMap
- taskInstances.values.map { getSystemStreamPartitionToTaskInstance }.flatten.groupBy(_._1).map {
- case (ssp, ssp2taskInstance) => ssp -> ssp2taskInstance.map(_._2).toList
- }
+ taskInstances.values
+ .flatMap(getSystemStreamPartitionToTaskInstance)
+ .groupBy(_._1)
+ .map { case (ssp, ssp2taskInstance) => ssp -> ssp2taskInstance.map(_._2).toList }
}
/**
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
index f4d605f..3adaeda 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
@@ -22,9 +22,7 @@ package org.apache.samza.container
import java.io.File
import java.nio.file.Path
import java.util
-import java.util.concurrent.ExecutorService
-import java.util.concurrent.Executors
-import java.util.concurrent.TimeUnit
+import java.util.concurrent.{CountDownLatch, ExecutorService, Executors, TimeUnit}
import java.lang.Thread.UncaughtExceptionHandler
import java.net.{URL, UnknownHostException}
import org.apache.samza.SamzaException
@@ -69,7 +67,9 @@ import org.apache.samza.system.chooser.RoundRobinChooserFactory
import org.apache.samza.task.AsyncRunLoop
import org.apache.samza.task.AsyncStreamTask
import org.apache.samza.task.AsyncStreamTaskAdapter
+import org.apache.samza.task.AsyncStreamTaskFactory
import org.apache.samza.task.StreamTask
+import org.apache.samza.task.StreamTaskFactory
import org.apache.samza.task.TaskInstanceCollector
import org.apache.samza.util.HighResolutionClock
import org.apache.samza.util.ExponentialSleepStrategy
@@ -165,7 +165,8 @@ object SamzaContainer extends Logging {
maxChangeLogStreamPartitions: Int,
localityManager: LocalityManager,
jmxServer: JmxServer,
- customReporters: Map[String, MetricsReporter] = Map[String, MetricsReporter]()) = {
+ customReporters: Map[String, MetricsReporter] = Map[String, MetricsReporter](),
+ taskFactory: Object = null) = {
val containerName = getSamzaContainerName(containerId)
val containerPID = Util.getContainerPID
@@ -234,12 +235,6 @@ object SamzaContainer extends Logging {
info("Got input stream metadata: %s" format inputStreamMetadata)
- val taskClassName = config
- .getTaskClass
- .getOrElse(throw new SamzaException("No task class defined in configuration."))
-
- info("Got stream task class: %s" format taskClassName)
-
val consumers = inputSystems
.map(systemName => {
val systemFactory = systemFactories(systemName)
@@ -248,7 +243,7 @@ object SamzaContainer extends Logging {
(systemName, systemFactory.getConsumer(systemName, config, samzaContainerMetrics.registry))
} catch {
case e: Exception =>
- error("Failed to create a consumer for %s, so skipping." format(systemName), e)
+ error("Failed to create a consumer for %s, so skipping." format systemName, e)
(systemName, null)
}
})
@@ -257,11 +252,6 @@ object SamzaContainer extends Logging {
info("Got system consumers: %s" format consumers.keys)
- val isAsyncTask = classOf[AsyncStreamTask].isAssignableFrom(Class.forName(taskClassName))
- if (isAsyncTask) {
- info("%s is AsyncStreamTask" format taskClassName)
- }
-
val producers = systemFactories
.map {
case (systemName, systemFactory) =>
@@ -269,12 +259,11 @@ object SamzaContainer extends Logging {
(systemName, systemFactory.getProducer(systemName, config, samzaContainerMetrics.registry))
} catch {
case e: Exception =>
- error("Failed to create a producer for %s, so skipping." format(systemName), e)
+ error("Failed to create a producer for %s, so skipping." format systemName, e)
(systemName, null)
}
}
.filter(_._2 != null)
- .toMap
info("Got system producers: %s" format producers.keys)
@@ -292,44 +281,48 @@ object SamzaContainer extends Logging {
info("Got serdes: %s" format serdes.keys)
/*
- * A Helper function to build a Map[String, Serde] (systemName -> Serde) for systems defined in the config. This is useful to build both key and message serde maps.
+ * A Helper function to build a Map[String, Serde] (systemName -> Serde) for systems defined
+ * in the config. This is useful to build both key and message serde maps.
*/
val buildSystemSerdeMap = (getSerdeName: (String) => Option[String]) => {
systemNames
.filter(systemName => getSerdeName(systemName).isDefined)
.map(systemName => {
val serdeName = getSerdeName(systemName).get
- val serde = serdes.getOrElse(serdeName, throw new SamzaException("buildSystemSerdeMap: No class defined for serde: %s." format serdeName))
+ val serde = serdes.getOrElse(serdeName,
+ throw new SamzaException("buildSystemSerdeMap: No class defined for serde: %s." format serdeName))
(systemName, serde)
}).toMap
}
/*
- * A Helper function to build a Map[SystemStream, Serde] for streams defined in the config. This is useful to build both key and message serde maps.
+ * A Helper function to build a Map[SystemStream, Serde] for streams defined in the config.
+ * This is useful to build both key and message serde maps.
*/
val buildSystemStreamSerdeMap = (getSerdeName: (SystemStream) => Option[String]) => {
(serdeStreams ++ inputSystemStreamPartitions)
.filter(systemStream => getSerdeName(systemStream).isDefined)
.map(systemStream => {
val serdeName = getSerdeName(systemStream).get
- val serde = serdes.getOrElse(serdeName, throw new SamzaException("buildSystemStreamSerdeMap: No class defined for serde: %s." format serdeName))
+ val serde = serdes.getOrElse(serdeName,
+ throw new SamzaException("buildSystemStreamSerdeMap: No class defined for serde: %s." format serdeName))
(systemStream, serde)
}).toMap
}
- val systemKeySerdes = buildSystemSerdeMap((systemName: String) => config.getSystemKeySerde(systemName))
+ val systemKeySerdes = buildSystemSerdeMap(systemName => config.getSystemKeySerde(systemName))
debug("Got system key serdes: %s" format systemKeySerdes)
- val systemMessageSerdes = buildSystemSerdeMap((systemName: String) => config.getSystemMsgSerde(systemName))
+ val systemMessageSerdes = buildSystemSerdeMap(systemName => config.getSystemMsgSerde(systemName))
debug("Got system message serdes: %s" format systemMessageSerdes)
- val systemStreamKeySerdes = buildSystemStreamSerdeMap((systemStream: SystemStream) => config.getStreamKeySerde(systemStream))
+ val systemStreamKeySerdes = buildSystemStreamSerdeMap(systemStream => config.getStreamKeySerde(systemStream))
debug("Got system stream key serdes: %s" format systemStreamKeySerdes)
- val systemStreamMessageSerdes = buildSystemStreamSerdeMap((systemStream: SystemStream) => config.getStreamMsgSerde(systemStream))
+ val systemStreamMessageSerdes = buildSystemStreamSerdeMap(systemStream => config.getStreamMsgSerde(systemStream))
debug("Got system stream message serdes: %s" format systemStreamMessageSerdes)
@@ -378,13 +371,10 @@ object SamzaContainer extends Logging {
val coordinatorSystemProducer = new CoordinatorStreamSystemFactory().getCoordinatorStreamSystemProducer(config, samzaContainerMetrics.registry)
val localityManager = new LocalityManager(coordinatorSystemProducer)
- val checkpointManager = config.getCheckpointManagerFactory() match {
- case Some(checkpointFactoryClassName) if (!checkpointFactoryClassName.isEmpty) =>
- Util
- .getObj[CheckpointManagerFactory](checkpointFactoryClassName)
- .getCheckpointManager(config, samzaContainerMetrics.registry)
- case _ => null
- }
+ val checkpointManager = config.getCheckpointManagerFactory()
+ .filterNot(_.isEmpty)
+ .map(Util.getObj[CheckpointManagerFactory](_).getCheckpointManager(config, samzaContainerMetrics.registry))
+ .orNull
info("Got checkpoint manager: %s" format checkpointManager)
// create a map of consumers with callbacks to pass to the OffsetManager
@@ -442,8 +432,26 @@ object SamzaContainer extends Logging {
val singleThreadMode = config.getSingleThreadMode
info("Got single thread mode: " + singleThreadMode)
+ val taskClassName = config.getTaskClass.orNull
+ info("Got task class name: %s" format taskClassName)
+
+ if (taskClassName == null && taskFactory == null) {
+ throw new SamzaException("Either the task class name or the task factory instance is required.")
+ }
+
+ val isAsyncTask: Boolean =
+ if (taskFactory != null) {
+ taskFactory.isInstanceOf[AsyncStreamTaskFactory]
+ } else {
+ classOf[AsyncStreamTask].isAssignableFrom(Class.forName(taskClassName))
+ }
+
+ if (isAsyncTask) {
+ info("Got an AsyncStreamTask implementation.")
+ }
+
if(singleThreadMode && isAsyncTask) {
- throw new SamzaException("AsyncStreamTask %s cannot run on single thread mode." format taskClassName)
+ throw new SamzaException("AsyncStreamTask cannot run on single thread mode.")
}
val threadPoolSize = config.getThreadPoolSize
@@ -470,12 +478,23 @@ object SamzaContainer extends Logging {
val storeWatchPaths = new util.HashSet[Path]()
storeWatchPaths.add(defaultStoreBaseDir.toPath)
- val taskInstances: Map[TaskName, TaskInstance[_]] = containerModel.getTasks.values.map(taskModel => {
+ val taskInstances: Map[TaskName, TaskInstance] = containerModel.getTasks.values.map(taskModel => {
debug("Setting up task instance: %s" format taskModel)
val taskName = taskModel.getTaskName
- val taskObj = Class.forName(taskClassName).newInstance
+ val taskObj = if (taskFactory != null) {
+ debug("Using task factory to create task instance")
+ taskFactory match {
+ case tf: AsyncStreamTaskFactory => tf.createInstance()
+ case tf: StreamTaskFactory => tf.createInstance()
+ case _ =>
+ throw new SamzaException("taskFactory must be an instance of StreamTaskFactory or AsyncStreamTaskFactory")
+ }
+ } else {
+ debug("Using task class name: %s to create instance" format taskClassName)
+ Class.forName(taskClassName).newInstance
+ }
val task = if (!singleThreadMode && !isAsyncTask)
// Wrap the StreamTask into a AsyncStreamTask with the build-in thread pool
@@ -491,18 +510,21 @@ object SamzaContainer extends Logging {
.map {
case (storeName, changeLogSystemStream) =>
val systemConsumer = systemFactories
- .getOrElse(changeLogSystemStream.getSystem, throw new SamzaException("Changelog system %s for store %s does not exist in the config." format (changeLogSystemStream, storeName)))
+ .getOrElse(changeLogSystemStream.getSystem,
+ throw new SamzaException("Changelog system %s for store %s does not " +
+ "exist in the config." format (changeLogSystemStream, storeName)))
.getConsumer(changeLogSystemStream.getSystem, config, taskInstanceMetrics.registry)
samzaContainerMetrics.addStoreRestorationGauge(taskName, storeName)
(storeName, systemConsumer)
- }.toMap
+ }
info("Got store consumers: %s" format storeConsumers)
var loggedStorageBaseDir: File = null
if(System.getenv(ShellCommandConfig.ENV_LOGGED_STORE_BASE_DIR) != null) {
val jobNameAndId = Util.getJobNameAndId(config)
- loggedStorageBaseDir = new File(System.getenv(ShellCommandConfig.ENV_LOGGED_STORE_BASE_DIR) + File.separator + jobNameAndId._1 + "-" + jobNameAndId._2)
+ loggedStorageBaseDir = new File(System.getenv(ShellCommandConfig.ENV_LOGGED_STORE_BASE_DIR)
+ + File.separator + jobNameAndId._1 + "-" + jobNameAndId._2)
} else {
warn("No override was provided for logged store base directory. This disables local state re-use on " +
"application restart. If you want to enable this feature, set LOGGED_STORE_BASE_DIR as an environment " +
@@ -523,11 +545,13 @@ object SamzaContainer extends Logging {
null
}
val keySerde = config.getStorageKeySerde(storeName) match {
- case Some(keySerde) => serdes.getOrElse(keySerde, throw new SamzaException("StorageKeySerde: No class defined for serde: %s." format keySerde))
+ case Some(keySerde) => serdes.getOrElse(keySerde,
+ throw new SamzaException("StorageKeySerde: No class defined for serde: %s." format keySerde))
case _ => null
}
val msgSerde = config.getStorageMsgSerde(storeName) match {
- case Some(msgSerde) => serdes.getOrElse(msgSerde, throw new SamzaException("StorageMsgSerde: No class defined for serde: %s." format msgSerde))
+ case Some(msgSerde) => serdes.getOrElse(msgSerde,
+ throw new SamzaException("StorageMsgSerde: No class defined for serde: %s." format msgSerde))
case _ => null
}
val storeBaseDir = if(changeLogSystemStreamPartition != null) {
@@ -568,7 +592,7 @@ object SamzaContainer extends Logging {
info("Retrieved SystemStreamPartitions " + systemStreamPartitions + " for " + taskName)
- def createTaskInstance[T] (task: T ): TaskInstance[T] = new TaskInstance[T](
+ def createTaskInstance(task: Any): TaskInstance = new TaskInstance(
task = task,
taskName = taskName,
config = config,
@@ -659,7 +683,7 @@ object SamzaContainer extends Logging {
class SamzaContainer(
containerContext: SamzaContainerContext,
- taskInstances: Map[TaskName, TaskInstance[_]],
+ taskInstances: Map[TaskName, TaskInstance],
runLoop: Runnable,
consumerMultiplexer: SystemConsumers,
producerMultiplexer: SystemProducers,
@@ -675,6 +699,17 @@ class SamzaContainer(
taskThreadPool: ExecutorService = null) extends Runnable with Logging {
val shutdownMs = containerContext.config.getShutdownMs.getOrElse(5000L)
+ private val runLoopStartLatch: CountDownLatch = new CountDownLatch(1)
+
+ def awaitStart(timeoutMs: Long): Boolean = {
+ try {
+ runLoopStartLatch.await(timeoutMs, TimeUnit.MILLISECONDS)
+ } catch {
+ case ie: InterruptedException =>
+ error("Interrupted while waiting for runloop to start!", ie)
+ throw ie
+ }
+ }
def run {
try {
@@ -691,8 +726,9 @@ class SamzaContainer(
startConsumers
startSecurityManger
- info("Entering run loop.")
addShutdownHook
+ runLoopStartLatch.countDown()
+ info("Entering run loop.")
runLoop.run
} catch {
case e: Exception =>
@@ -716,6 +752,13 @@ class SamzaContainer(
}
}
+ def shutdown() = {
+ runLoop match {
+ case runLoop: RunLoop => runLoop.shutdown
+ case asyncRunLoop: AsyncRunLoop => asyncRunLoop.shutdown()
+ }
+ }
+
def startDiskSpaceMonitor: Unit = {
if (diskSpaceMonitor != null) {
info("Starting disk space monitor")
@@ -773,9 +816,11 @@ class SamzaContainer(
localityManager.writeContainerToHostMapping(containerContext.id, hostInet.getHostName, jmxUrl, jmxTunnelingUrl)
} catch {
case uhe: UnknownHostException =>
- warn("Received UnknownHostException when persisting locality info for container %d: %s" format (containerContext.id, uhe.getMessage)) //No-op
+ warn("Received UnknownHostException when persisting locality info for container %d: " +
+ "%s" format (containerContext.id, uhe.getMessage)) //No-op
case unknownException: Throwable =>
- warn("Received an exception when persisting locality info for container %d: %s" format (containerContext.id, unknownException.getMessage))
+ warn("Received an exception when persisting locality info for container %d: " +
+ "%s" format (containerContext.id, unknownException.getMessage))
}
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala b/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
index 26a8f5f..e07fcf4 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
@@ -43,8 +43,8 @@ import org.apache.samza.util.Logging
import scala.collection.JavaConversions._
-class TaskInstance[T](
- task: T,
+class TaskInstance(
+ task: Any,
val taskName: TaskName,
config: Config,
val metrics: TaskInstanceMetrics,
@@ -84,7 +84,8 @@ class TaskInstance[T](
// store the (ssp -> if this ssp is catched up) mapping. "catched up"
// means the same ssp in other taskInstances have the same offset as
// the one here.
- var ssp2catchedupMapping: scala.collection.mutable.Map[SystemStreamPartition, Boolean] = scala.collection.mutable.Map[SystemStreamPartition, Boolean]()
+ var ssp2catchedupMapping: scala.collection.mutable.Map[SystemStreamPartition, Boolean] =
+ scala.collection.mutable.Map[SystemStreamPartition, Boolean]()
systemStreamPartitions.foreach(ssp2catchedupMapping += _ -> false)
def registerMetrics {
@@ -140,7 +141,8 @@ class TaskInstance[T](
})
}
- def process(envelope: IncomingMessageEnvelope, coordinator: ReadableCoordinator, callbackFactory: TaskCallbackFactory = null) {
+ def process(envelope: IncomingMessageEnvelope, coordinator: ReadableCoordinator,
+ callbackFactory: TaskCallbackFactory = null) {
metrics.processes.inc
if (!ssp2catchedupMapping.getOrElse(envelope.getSystemStreamPartition,
@@ -151,7 +153,8 @@ class TaskInstance[T](
if (ssp2catchedupMapping(envelope.getSystemStreamPartition)) {
metrics.messagesActuallyProcessed.inc
- trace("Processing incoming message envelope for taskName and SSP: %s, %s" format (taskName, envelope.getSystemStreamPartition))
+ trace("Processing incoming message envelope for taskName and SSP: %s, %s"
+ format (taskName, envelope.getSystemStreamPartition))
if (isAsyncTask) {
exceptionHandler.maybeHandle {
@@ -163,7 +166,8 @@ class TaskInstance[T](
task.asInstanceOf[StreamTask].process(envelope, collector, coordinator)
}
- trace("Updating offset map for taskName, SSP and offset: %s, %s, %s" format (taskName, envelope.getSystemStreamPartition, envelope.getOffset))
+ trace("Updating offset map for taskName, SSP and offset: %s, %s, %s"
+ format (taskName, envelope.getSystemStreamPartition, envelope.getOffset))
offsetManager.update(taskName, envelope.getSystemStreamPartition, envelope.getOffset)
}
@@ -173,7 +177,7 @@ class TaskInstance[T](
def endOfStream(coordinator: ReadableCoordinator): Unit = {
if (isEndOfStreamListenerTask) {
exceptionHandler.maybeHandle {
- task.asInstanceOf[EndOfStreamListenerTask].onEndOfStream(collector, coordinator);
+ task.asInstanceOf[EndOfStreamListenerTask].onEndOfStream(collector, coordinator)
}
}
}
@@ -230,7 +234,8 @@ class TaskInstance[T](
override def toString() = "TaskInstance for class %s and taskName %s." format (task.getClass.getName, taskName)
- def toDetailedString() = "TaskInstance [taskName = %s, windowable=%s, closable=%s endofstreamlistener=%s]" format (taskName, isWindowableTask, isClosableTask, isEndOfStreamListenerTask)
+ def toDetailedString() = "TaskInstance [taskName = %s, windowable=%s, closable=%s endofstreamlistener=%s]" format
+ (taskName, isWindowableTask, isClosableTask, isEndOfStreamListenerTask)
/**
* From the envelope, check if this SSP has catched up with the starting offset of the SSP
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala b/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala
index 9ccf6fc..0d5adee 100644
--- a/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala
+++ b/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala
@@ -49,22 +49,14 @@ class ThreadJobFactory extends StreamJobFactory with Logging {
try {
coordinator.start
- new ThreadJob(new Runnable {
- override def run(): Unit = {
- val jmxServer = new JmxServer
- try {
+ new ThreadJob(
SamzaContainer(
containerModel.getContainerId,
containerModel,
config,
jobModel.maxChangeLogStreamPartitions,
null,
- new JmxServer)
- } finally {
- jmxServer.stop
- }
- }
- })
+ new JmxServer))
} finally {
coordinator.stop
}
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/samza-core/src/test/java/org/apache/samza/task/TestAsyncRunLoop.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/task/TestAsyncRunLoop.java b/samza-core/src/test/java/org/apache/samza/task/TestAsyncRunLoop.java
index 58975d3..798977b 100644
--- a/samza-core/src/test/java/org/apache/samza/task/TestAsyncRunLoop.java
+++ b/samza-core/src/test/java/org/apache/samza/task/TestAsyncRunLoop.java
@@ -60,7 +60,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
public class TestAsyncRunLoop {
- Map<TaskName, TaskInstance<AsyncStreamTask>> tasks;
+ Map<TaskName, TaskInstance> tasks;
ExecutorService executor;
SystemConsumers consumerMultiplexer;
SamzaContainerMetrics containerMetrics;
@@ -87,8 +87,8 @@ public class TestAsyncRunLoop {
TestTask task0;
TestTask task1;
- TaskInstance<AsyncStreamTask> t0;
- TaskInstance<AsyncStreamTask> t1;
+ TaskInstance t0;
+ TaskInstance t1;
AsyncRunLoop createRunLoop() {
return new AsyncRunLoop(tasks,
@@ -103,15 +103,15 @@ public class TestAsyncRunLoop {
() -> 0L);
}
- TaskInstance<AsyncStreamTask> createTaskInstance(AsyncStreamTask task, TaskName taskName, SystemStreamPartition ssp, OffsetManager manager, SystemConsumers consumers) {
+ TaskInstance createTaskInstance(AsyncStreamTask task, TaskName taskName, SystemStreamPartition ssp, OffsetManager manager, SystemConsumers consumers) {
TaskInstanceMetrics taskInstanceMetrics = new TaskInstanceMetrics("task", new MetricsRegistryMap());
scala.collection.immutable.Set<SystemStreamPartition> sspSet = JavaConversions.asScalaSet(Collections.singleton(ssp)).toSet();
- return new TaskInstance<AsyncStreamTask>(task, taskName, mock(Config.class), taskInstanceMetrics,
+ return new TaskInstance(task, taskName, mock(Config.class), taskInstanceMetrics,
null, consumers, mock(TaskInstanceCollector.class), mock(SamzaContainerContext.class),
manager, null, null, sspSet, new TaskInstanceExceptionHandler(taskInstanceMetrics, new scala.collection.immutable.HashSet<String>()));
}
- TaskInstance<AsyncStreamTask> createTaskInstance(AsyncStreamTask task, TaskName taskName, SystemStreamPartition ssp) {
+ TaskInstance createTaskInstance(AsyncStreamTask task, TaskName taskName, SystemStreamPartition ssp) {
return createTaskInstance(task, taskName, ssp, offsetManager, consumerMultiplexer);
}
@@ -466,7 +466,7 @@ public class TestAsyncRunLoop {
sspMap.put(ssp2, messageList);
SystemConsumer mockConsumer = mock(SystemConsumer.class);
- when(mockConsumer.poll((Set<SystemStreamPartition>) anyObject(), anyLong())).thenReturn(sspMap);
+ when(mockConsumer.poll(anyObject(), anyLong())).thenReturn(sspMap);
HashMap<String, SystemConsumer> systemConsumerMap = new HashMap<>();
systemConsumerMap.put("system1", mockConsumer);
@@ -485,9 +485,9 @@ public class TestAsyncRunLoop {
when(offsetManager.getStartingOffset(taskName1, ssp1)).thenReturn(Option.apply(IncomingMessageEnvelope.END_OF_STREAM_OFFSET));
when(offsetManager.getStartingOffset(taskName2, ssp2)).thenReturn(Option.apply("1"));
- TaskInstance<AsyncStreamTask> taskInstance1 = createTaskInstance(mockStreamTask1, taskName1, ssp1, offsetManager, consumers);
- TaskInstance<AsyncStreamTask> taskInstance2 = createTaskInstance(mockStreamTask2, taskName2, ssp2, offsetManager, consumers);
- Map<TaskName, TaskInstance<AsyncStreamTask>> tasks = new HashMap<>();
+ TaskInstance taskInstance1 = createTaskInstance(mockStreamTask1, taskName1, ssp1, offsetManager, consumers);
+ TaskInstance taskInstance2 = createTaskInstance(mockStreamTask2, taskName2, ssp2, offsetManager, consumers);
+ Map<TaskName, TaskInstance> tasks = new HashMap<>();
tasks.put(taskName1, taskInstance1);
tasks.put(taskName2, taskInstance2);
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala b/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala
index d83c7e2..0304fc8 100644
--- a/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala
+++ b/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala
@@ -31,7 +31,6 @@ import org.apache.samza.system.SystemConsumers
import org.apache.samza.system.SystemStreamPartition
import org.apache.samza.task.TaskCoordinator.RequestScope
import org.apache.samza.task.ReadableCoordinator
-import org.apache.samza.task.StreamTask
import org.apache.samza.util.Clock
import org.junit.Assert._
import org.junit.Test
@@ -55,12 +54,12 @@ class TestRunLoop extends AssertionsForJUnit with MockitoSugar with ScalaTestMat
val envelope0 = new IncomingMessageEnvelope(ssp0, "0", "key0", "value0")
val envelope1 = new IncomingMessageEnvelope(ssp1, "1", "key1", "value1")
- def getMockTaskInstances: Map[TaskName, TaskInstance[StreamTask]] = {
- val ti0 = mock[TaskInstance[StreamTask]]
+ def getMockTaskInstances: Map[TaskName, TaskInstance] = {
+ val ti0 = mock[TaskInstance]
when(ti0.systemStreamPartitions).thenReturn(Set(ssp0))
when(ti0.taskName).thenReturn(taskName0)
- val ti1 = mock[TaskInstance[StreamTask]]
+ val ti1 = mock[TaskInstance]
when(ti1.systemStreamPartitions).thenReturn(Set(ssp1))
when(ti1.taskName).thenReturn(taskName1)
@@ -183,7 +182,7 @@ class TestRunLoop extends AssertionsForJUnit with MockitoSugar with ScalaTestMat
def anyObject[T] = Matchers.anyObject.asInstanceOf[T]
// Stub out TaskInstance.process. Mockito really doesn't make this easy. :(
- def stubProcess(taskInstance: TaskInstance[StreamTask], process: (IncomingMessageEnvelope, ReadableCoordinator) => Unit) {
+ def stubProcess(taskInstance: TaskInstance, process: (IncomingMessageEnvelope, ReadableCoordinator) => Unit) {
when(taskInstance.process(anyObject, anyObject, anyObject)).thenAnswer(new Answer[Unit]() {
override def answer(invocation: InvocationOnMock) {
val envelope = invocation.getArguments()(0).asInstanceOf[IncomingMessageEnvelope]
@@ -276,9 +275,9 @@ class TestRunLoop extends AssertionsForJUnit with MockitoSugar with ScalaTestMat
@Test
def testGetSystemStreamPartitionToTaskInstancesMapping {
- val ti0 = mock[TaskInstance[StreamTask]]
- val ti1 = mock[TaskInstance[StreamTask]]
- val ti2 = mock[TaskInstance[StreamTask]]
+ val ti0 = mock[TaskInstance]
+ val ti1 = mock[TaskInstance]
+ val ti2 = mock[TaskInstance]
when(ti0.systemStreamPartitions).thenReturn(Set(ssp0))
when(ti1.systemStreamPartitions).thenReturn(Set(ssp1))
when(ti2.systemStreamPartitions).thenReturn(Set(ssp1))
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala b/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala
index 5895037..07055a0 100644
--- a/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala
+++ b/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala
@@ -180,7 +180,7 @@ class TestSamzaContainer extends AssertionsForJUnit with MockitoSugar {
new SerdeManager)
val collector = new TaskInstanceCollector(producerMultiplexer)
val containerContext = new SamzaContainerContext(0, config, Set[TaskName](taskName))
- val taskInstance: TaskInstance[StreamTask] = new TaskInstance[StreamTask](
+ val taskInstance: TaskInstance = new TaskInstance(
task,
taskName,
config,
@@ -262,7 +262,7 @@ class TestSamzaContainer extends AssertionsForJUnit with MockitoSugar {
}
})
- val taskInstance: TaskInstance[StreamTask] = new TaskInstance[StreamTask](
+ val taskInstance: TaskInstance = new TaskInstance(
task,
taskName,
config,
http://git-wip-us.apache.org/repos/asf/samza/blob/be6be8cb/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala b/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
index 3c83529..7e35525 100644
--- a/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
+++ b/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
@@ -71,7 +71,7 @@ class TestTaskInstance {
val taskName = new TaskName("taskName")
val collector = new TaskInstanceCollector(producerMultiplexer)
val containerContext = new SamzaContainerContext(0, config, Set[TaskName](taskName))
- val taskInstance: TaskInstance[StreamTask] = new TaskInstance[StreamTask](
+ val taskInstance: TaskInstance = new TaskInstance(
task,
taskName,
config,
@@ -169,7 +169,7 @@ class TestTaskInstance {
val registry = new MetricsRegistryMap
val taskMetrics = new TaskInstanceMetrics(registry = registry)
- val taskInstance: TaskInstance[StreamTask] = new TaskInstance[StreamTask](
+ val taskInstance = new TaskInstance(
task,
taskName,
config,
@@ -226,7 +226,7 @@ class TestTaskInstance {
val registry = new MetricsRegistryMap
val taskMetrics = new TaskInstanceMetrics(registry = registry)
- val taskInstance: TaskInstance[StreamTask] = new TaskInstance[StreamTask](
+ val taskInstance = new TaskInstance(
task,
taskName,
config,
[3/7] samza git commit: job model generation cleanup
Posted by na...@apache.org.
job model generation cleanup
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/f15b9071
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/f15b9071
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/f15b9071
Branch: refs/heads/samza-standalone
Commit: f15b907127774bfb38f753a906b563560c6f657a
Parents: be6be8c
Author: navina <na...@apache.org>
Authored: Fri Dec 23 17:00:04 2016 -0800
Committer: navina <na...@apache.org>
Committed: Fri Dec 23 17:00:04 2016 -0800
----------------------------------------------------------------------
.../grouper/task/BalancingTaskNameGrouper.java | 9 ++++++++-
.../main/java/org/apache/samza/zk/ZkJobCoordinator.java | 12 +++++++-----
2 files changed, 15 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/f15b9071/samza-core/src/main/java/org/apache/samza/container/grouper/task/BalancingTaskNameGrouper.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/container/grouper/task/BalancingTaskNameGrouper.java b/samza-core/src/main/java/org/apache/samza/container/grouper/task/BalancingTaskNameGrouper.java
index f8295c8..1946271 100644
--- a/samza-core/src/main/java/org/apache/samza/container/grouper/task/BalancingTaskNameGrouper.java
+++ b/samza-core/src/main/java/org/apache/samza/container/grouper/task/BalancingTaskNameGrouper.java
@@ -18,6 +18,7 @@
*/
package org.apache.samza.container.grouper.task;
+import java.util.Collections;
import java.util.Set;
import org.apache.samza.container.LocalityManager;
import org.apache.samza.job.model.ContainerModel;
@@ -54,5 +55,11 @@ public interface BalancingTaskNameGrouper extends TaskNameGrouper {
* @param localityManager provides a persisted task to container map to use as a baseline
* @return the grouped tasks in the form of ContainerModels
*/
- Set<ContainerModel> balance(Set<TaskModel> tasks, LocalityManager localityManager);
+ default Set<ContainerModel> balance(Set<TaskModel> tasks, LocalityManager localityManager) {
+ return Collections.<ContainerModel>emptySet();
+ }
+
+ default Set<ContainerModel> balance(Set<Integer> containerIds, Set<TaskModel> tasks, LocalityManager localityManager) {
+ return Collections.<ContainerModel>emptySet();
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/f15b9071/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 8c36ff2..3014c1b 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
@@ -8,6 +8,8 @@ import org.apache.samza.config.Config;
import org.apache.samza.config.JavaSystemConfig;
import org.apache.samza.config.MapConfig;
import org.apache.samza.coordinator.JobCoordinator;
+import org.apache.samza.coordinator.JobModelManager;
+import org.apache.samza.coordinator.JobModelManager$;
import org.apache.samza.job.model.ContainerModel;
import org.apache.samza.job.model.JobModel;
import org.apache.samza.processor.SamzaContainerController;
@@ -41,7 +43,7 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
private Config config;
private ZkKeyBuilder keyBuilder;
private final ScheduleAfterDebounceTime debounceTimer;
- //JobModelManager jobModelManager;
+ JobModelManager jobModelManager;
public ZkJobCoordinator(int processorId, Config config, ScheduleAfterDebounceTime debounceTimer, ZkUtils zkUtils, SamzaContainerController containerController) {
this.zkUtils = zkUtils;
@@ -75,7 +77,7 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
streamMetadataCache = new StreamMetadataCache(Util.<String, SystemAdmin>javaMapAsScalaMap(systemAdmins), 5000, SystemClock
.instance());
- //jobModelManager = //JobModelManager$.MODULE$.getJobCoordinator(this.config, null, null, streamMetadataCache, null);
+ jobModelManager = JobModelManager$.MODULE$.getJobCoordinator(this.config, null, null, streamMetadataCache, null);
////////////////////////////////////////////////////////////////////////////////////////////
}
@@ -132,7 +134,8 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
Map<String, String> configMap = new HashMap<>();
Map<Integer, ContainerModel> containers = new HashMap<>();
MapConfig config = new MapConfig(configMap);
- JobModel jobModel = new JobModel(config, containers);
+ //JobModel jobModel = new JobModel(config, containers);
+ JobModel jobModel = jobModelManager.jobModel();
log.info("pid=" + processorId + "Generated jobModel: " + jobModel);
@@ -188,11 +191,10 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
// ?????
JobModel jobModel = getJobModel();
log.info("pid=" + processorId + "got the new job model =" + jobModel);
- /*
+
containerController.startContainer(
jobModel.getContainers().get(processorId),
jobModel.getConfig(),
jobModel.maxChangeLogStreamPartitions);
- */
}
}
[7/7] samza git commit: Fixing checkstyle errors
Posted by na...@apache.org.
Fixing checkstyle errors
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/4918e3ad
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/4918e3ad
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/4918e3ad
Branch: refs/heads/samza-standalone
Commit: 4918e3ad7c622a36f6c8d8dc8eb6c7778fdaac05
Parents: 0a2b63a
Author: navina <na...@apache.org>
Authored: Fri Dec 23 17:03:25 2016 -0800
Committer: navina <na...@apache.org>
Committed: Fri Dec 23 17:03:25 2016 -0800
----------------------------------------------------------------------
.../org/apache/samza/config/JavaJobConfig.java | 21 ++++++++-
.../samza/config/JobCoordinatorConfig.java | 23 ++++++++--
.../java/org/apache/samza/config/ZkConfig.java | 21 ++++++++-
.../task/SimpleGroupByContainerCount.java | 30 ++++++++++---
.../SimpleGroupByContainerCountFactory.java | 1 -
.../leaderelection/LeaderElector.java | 19 ++++++++
.../processor/SamzaContainerController.java | 41 ++++++++++++-----
.../samza/zk/BarrierForVersionUpgrade.java | 19 ++++++++
.../samza/zk/ScheduleAfterDebounceTime.java | 34 ++++++++++++---
.../samza/zk/ZkBarrierForVersionUpgrade.java | 39 +++++++++++++----
.../java/org/apache/samza/zk/ZkController.java | 27 +++++++++++-
.../org/apache/samza/zk/ZkControllerImpl.java | 36 ++++++++++++---
.../org/apache/samza/zk/ZkJobCoordinator.java | 46 +++++++++++++-------
.../samza/zk/ZkJobCoordinatorFactory.java | 19 ++++++++
.../java/org/apache/samza/zk/ZkKeyBuilder.java | 28 ++++++++++--
.../org/apache/samza/zk/ZkLeaderElector.java | 23 +++++++++-
.../java/org/apache/samza/zk/ZkListener.java | 19 ++++++++
.../main/java/org/apache/samza/zk/ZkUtils.java | 46 +++++++++++++++-----
18 files changed, 413 insertions(+), 79 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/config/JavaJobConfig.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/config/JavaJobConfig.java b/samza-core/src/main/java/org/apache/samza/config/JavaJobConfig.java
index c0747f0..24b9427 100644
--- a/samza-core/src/main/java/org/apache/samza/config/JavaJobConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/JavaJobConfig.java
@@ -1,3 +1,22 @@
+/*
+ * 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.config;
public class JavaJobConfig extends MapConfig {
@@ -5,7 +24,7 @@ public class JavaJobConfig extends MapConfig {
private static final String JOB_ID = "job.id"; // streaming.job_id
private static final String DEFAULT_JOB_ID = "1";
- public JavaJobConfig (Config config) {
+ public JavaJobConfig(Config config) {
super(config);
}
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/config/JobCoordinatorConfig.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/config/JobCoordinatorConfig.java b/samza-core/src/main/java/org/apache/samza/config/JobCoordinatorConfig.java
index c8e496e..4bb66de 100644
--- a/samza-core/src/main/java/org/apache/samza/config/JobCoordinatorConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/JobCoordinatorConfig.java
@@ -1,14 +1,31 @@
+/*
+ * 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.config;
import com.google.common.base.Strings;
-import org.apache.samza.coordinator.JobCoordinatorFactory;
-import org.apache.samza.util.Util;
public class JobCoordinatorConfig extends MapConfig {
// TODO: Change this to job-coordinator.factory
private static final String JOB_COORDINATOR_FACTORY = "job.coordinator.factory";
- public JobCoordinatorConfig (Config config) {
+ public JobCoordinatorConfig(Config config) {
super(config);
}
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java b/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java
index 973db42..31b1eda 100644
--- a/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java
@@ -1,3 +1,22 @@
+/*
+ * 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.config;
public class ZkConfig extends MapConfig {
@@ -9,7 +28,7 @@ public class ZkConfig extends MapConfig {
public static final int DEFAULT_CONNECTION_TIMEOUT_MS = 60000;
public static final int DEFAULT_SESSION_TIMEOUT_MS = 30000;
- public ZkConfig (Config config) {
+ public ZkConfig(Config config) {
super(config);
}
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java b/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java
index 359c4ed..cf225c3 100644
--- a/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java
+++ b/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java
@@ -1,5 +1,28 @@
+/*
+ * 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.container.grouper.task;
+import org.apache.samza.container.TaskName;
+import org.apache.samza.job.model.ContainerModel;
+import org.apache.samza.job.model.TaskModel;
+
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -7,10 +30,6 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import org.apache.samza.container.LocalityManager;
-import org.apache.samza.container.TaskName;
-import org.apache.samza.job.model.ContainerModel;
-import org.apache.samza.job.model.TaskModel;
public class SimpleGroupByContainerCount implements TaskNameGrouper {
@@ -18,6 +37,7 @@ public class SimpleGroupByContainerCount implements TaskNameGrouper {
public SimpleGroupByContainerCount() {
this.startContainerCount = 1;
}
+
public SimpleGroupByContainerCount(int containerCount) {
if (containerCount <= 0) throw new IllegalArgumentException("Must have at least one container");
this.startContainerCount = containerCount;
@@ -33,7 +53,7 @@ public class SimpleGroupByContainerCount implements TaskNameGrouper {
}
public Set<ContainerModel> group(Set<TaskModel> tasks, List<Integer> containersIds) {
- if(containersIds == null)
+ if (containersIds == null)
return this.group(tasks);
int containerCount = containersIds.size();
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCountFactory.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCountFactory.java b/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCountFactory.java
index 02918f6..b1c6b92 100644
--- a/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCountFactory.java
+++ b/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCountFactory.java
@@ -20,7 +20,6 @@ package org.apache.samza.container.grouper.task;
import org.apache.samza.config.Config;
-import org.apache.samza.config.JobConfig;
/**
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/coordinator/leaderelection/LeaderElector.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/coordinator/leaderelection/LeaderElector.java b/samza-core/src/main/java/org/apache/samza/coordinator/leaderelection/LeaderElector.java
index fc3cac9..2565ee2 100644
--- a/samza-core/src/main/java/org/apache/samza/coordinator/leaderelection/LeaderElector.java
+++ b/samza-core/src/main/java/org/apache/samza/coordinator/leaderelection/LeaderElector.java
@@ -1,3 +1,22 @@
+/*
+ * 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.coordinator.leaderelection;
public interface LeaderElector {
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java b/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java
index fb227d0..d448d30 100644
--- a/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java
+++ b/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java
@@ -1,3 +1,22 @@
+/*
+ * 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.processor;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -38,12 +57,12 @@ public class SamzaContainerController {
* Creates an instance of a controller for instantiating, starting and/or stopping {@link SamzaContainer}
* Requests to execute a container are submitted to the {@link ExecutorService}
*
- * @param taskFactory Factory that be used create instances of {@link org.apache.samza.task.StreamTask} or
- * {@link org.apache.samza.task.AsyncStreamTask}
+ * @param taskFactory Factory that be used create instances of {@link org.apache.samza.task.StreamTask} or
+ * {@link org.apache.samza.task.AsyncStreamTask}
* @param containerShutdownMs How long the Samza container should wait for an orderly shutdown of task instances
- * @param metricsReporterMap Map of metric reporter name and {@link MetricsReporter} instance
+ * @param metricsReporterMap Map of metric reporter name and {@link MetricsReporter} instance
*/
- public SamzaContainerController (
+ public SamzaContainerController(
Object taskFactory,
long containerShutdownMs,
String processorId,
@@ -62,14 +81,14 @@ public class SamzaContainerController {
/**
* Instantiates a container and submits to the executor. This method does not actually wait for the container to
* fully start-up. For such a behavior, see {@link #awaitStart(long)}
- *
+ * <p>
* <b>Note:</b> <i>This method does not stop a currently running container, if any. It is left up to the caller to
* ensure that the container has been stopped with stopContainer before invoking this method.</i>
*
- * @param containerModel {@link ContainerModel} instance to use for the current run of the Container
- * @param config Complete configuration map used by the Samza job
+ * @param containerModel {@link ContainerModel} instance to use for the current run of the Container
+ * @param config Complete configuration map used by the Samza job
* @param maxChangelogStreamPartitions Max number of partitions expected in the changelog streams
- * TODO: Try to get rid of maxChangelogStreamPartitions from method arguments
+ * TODO: Try to get rid of maxChangelogStreamPartitions from method arguments
*/
public void startContainer(ContainerModel containerModel, Config config, int maxChangelogStreamPartitions) {
LocalityManager localityManager = null;
@@ -96,7 +115,7 @@ public class SamzaContainerController {
*
* @param timeoutMs Maximum time to wait, in milliseconds
* @return {@code true}, if the container started within the specified wait time and {@code false} if the waiting
- * time elapsed
+ * time elapsed
* @throws InterruptedException if the current thread is interrupted while waiting for container to start-up
*/
public boolean awaitStart(long timeoutMs) throws InterruptedException {
@@ -107,14 +126,14 @@ public class SamzaContainerController {
* Stops a running container, if any. Invoking this method multiple times does not have any side-effects.
*/
public void stopContainer() {
- if(container == null) {
+ if (container == null) {
log.warn("Shutdown before a container was created.");
return;
}
container.shutdown();
try {
- if(containerFuture != null)
+ if (containerFuture != null)
containerFuture.get(containerShutdownMs, TimeUnit.MILLISECONDS);
} catch (InterruptedException | ExecutionException e) {
log.error("Ran into problems while trying to stop the container in the processor!", e);
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/zk/BarrierForVersionUpgrade.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/BarrierForVersionUpgrade.java b/samza-core/src/main/java/org/apache/samza/zk/BarrierForVersionUpgrade.java
index 691aced..c2c4c23 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/BarrierForVersionUpgrade.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/BarrierForVersionUpgrade.java
@@ -1,3 +1,22 @@
+/*
+ * 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.zk;
import java.util.List;
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java b/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java
index e217dab..d174938 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java
@@ -1,18 +1,39 @@
+/*
+ * 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.zk;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ScheduleAfterDebounceTime {
public static final Logger LOG = LoggerFactory.getLogger(ScheduleAfterDebounceTime.class);
- public static final long timeoutMs = 1000*10;
+ public static final long TIMEOUT_MS = 1000 * 10;
public static final String JOB_MODEL_VERSION_CHANGE = "JobModelVersionChange";
public static final String ON_PROCESSOR_CHANGE = "OnProcessorChange";
@@ -24,19 +45,18 @@ public class ScheduleAfterDebounceTime {
new ThreadFactoryBuilder().setNameFormat("zk-debounce-thread-%d").setDaemon(true).build());
private final Map<String, ScheduledFuture> futureHandles = new HashMap<>();
- public ScheduleAfterDebounceTime () {
-
+ public ScheduleAfterDebounceTime() {
}
- synchronized public void scheduleAfterDebounceTime (String actionName, long debounceTimeMs, Runnable runnable) {//, final ReadyToCreateJobModelListener listener) {
+ synchronized public void scheduleAfterDebounceTime(String actionName, long debounceTimeMs, Runnable runnable) { //, final ReadyToCreateJobModelListener listener) {
// check if this action has been scheduled already
ScheduledFuture sf = futureHandles.get(actionName);
- if(sf != null && !sf.isDone()) {
+ if (sf != null && !sf.isDone()) {
LOG.info(">>>>>>>>>>>DEBOUNCE: cancel future for " + actionName);
// attempt to cancel
- if(! sf.cancel(false) ) {
+ if (!sf.cancel(false)) {
try {
- sf.get(timeoutMs, TimeUnit.MILLISECONDS);
+ sf.get(TIMEOUT_MS, TimeUnit.MILLISECONDS);
} catch (Exception e) {
// we ignore the exception
LOG.warn("cancel for action " + actionName + " failed with ", e);
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/zk/ZkBarrierForVersionUpgrade.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkBarrierForVersionUpgrade.java b/samza-core/src/main/java/org/apache/samza/zk/ZkBarrierForVersionUpgrade.java
index 60a06da..524afed 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkBarrierForVersionUpgrade.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkBarrierForVersionUpgrade.java
@@ -1,6 +1,26 @@
+/*
+ * 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.zk;
import java.util.List;
+
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.slf4j.Logger;
@@ -17,7 +37,7 @@ public class ZkBarrierForVersionUpgrade implements BarrierForVersionUpgrade {
final private String barrierPrefix;
- public ZkBarrierForVersionUpgrade( ZkUtils zkUtils, ScheduleAfterDebounceTime debounceTimer) {
+ public ZkBarrierForVersionUpgrade(ZkUtils zkUtils, ScheduleAfterDebounceTime debounceTimer) {
this.zkUtils = zkUtils;
keyBuilder = zkUtils.getKeyBuilder();
@@ -83,7 +103,7 @@ public class ZkBarrierForVersionUpgrade implements BarrierForVersionUpgrade {
// Find out the event & Log
boolean allIn = true;
- if(currentChildren == null) {
+ if (currentChildren == null) {
LOG.info("Got handleChildChange with null currentChildren");
return;
}
@@ -101,33 +121,34 @@ public class ZkBarrierForVersionUpgrade implements BarrierForVersionUpgrade {
// check if all the names are in
- for(String n : names) {
- if(!currentChildren.contains(n)) {
+ for (String n : names) {
+ if (!currentChildren.contains(n)) {
LOG.info("node " + n + " is still not in the list ");
allIn = false;
break;
}
}
- if(allIn) {
+ if (allIn) {
LOG.info("ALl nodes reached the barrier");
callback.run(); // all the names have registered
}
}
}
- class ZkBarrierReachedHandler implements IZkDataListener {
+ class ZkBarrierReachedHandler implements IZkDataListener {
private final ScheduleAfterDebounceTime debounceTimer;
private final String barrierPathDone;
private final Runnable callback;
+
public ZkBarrierReachedHandler(String barrierPathDone, ScheduleAfterDebounceTime debounceTimer, Runnable callback) {
this.barrierPathDone = barrierPathDone;
- this.callback = callback;
+ this.callback = callback;
this.debounceTimer = debounceTimer;
}
@Override
public void handleDataChange(String dataPath, Object data)
- throws Exception {
+ throws Exception {
String done = (String) data;
LOG.info("got notification about barrier path=" + barrierPathDone + "; done=" + done);
if (done.equals(BARRIER_DONE)) {
@@ -140,7 +161,7 @@ public class ZkBarrierForVersionUpgrade implements BarrierForVersionUpgrade {
@Override
public void handleDataDeleted(String dataPath)
- throws Exception {
+ throws Exception {
LOG.warn("barrier done got deleted at " + dataPath);
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/zk/ZkController.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkController.java b/samza-core/src/main/java/org/apache/samza/zk/ZkController.java
index 20e55ab..76ff8d2 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkController.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkController.java
@@ -1,11 +1,34 @@
-package org.apache.samza.zk;
+/*
+ * 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.zk;
public interface ZkController {
- void register ();
+ void register();
+
boolean isLeader();
+
void notifyJobModelChange(String version);
+
void stop();
+
void listenToProcessorLiveness();
+
String currentJobModelVersion();
}
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java b/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java
index fdd1f02..b00a3ee 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkControllerImpl.java
@@ -1,3 +1,22 @@
+/*
+ * 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.zk;
import org.I0Itec.zkclient.IZkChildListener;
@@ -18,7 +37,7 @@ public class ZkControllerImpl implements ZkController {
private final ZkLeaderElector leaderElector;
private final ScheduleAfterDebounceTime debounceTimer;
- public ZkControllerImpl (String processorIdStr, ZkUtils zkUtils, ScheduleAfterDebounceTime debounceTimer, ZkListener zkListener) {
+ public ZkControllerImpl(String processorIdStr, ZkUtils zkUtils, ScheduleAfterDebounceTime debounceTimer, ZkListener zkListener) {
this.processorIdStr = processorIdStr;
this.zkUtils = zkUtils;
this.zkListener = zkListener;
@@ -30,11 +49,10 @@ public class ZkControllerImpl implements ZkController {
@Override
public void register() {
-
// TODO - make a loop here with some number of attempts.
// possibly split into two method - becomeLeader() and becomeParticipant()
boolean isLeader = leaderElector.tryBecomeLeader();
- if(isLeader) {
+ if (isLeader) {
listenToProcessorLiveness();
// zkUtils.subscribeToProcessorChange(zkProcessorChangeListener);
@@ -49,7 +67,7 @@ public class ZkControllerImpl implements ZkController {
private void init() {
ZkKeyBuilder keyBuilder = zkUtils.getKeyBuilder();
- zkUtils.makeSurePersistentPathsExists(new String[] {
+ zkUtils.makeSurePersistentPathsExists(new String[]{
keyBuilder.getProcessorsPath(), keyBuilder.getJobModelVersionPath(), keyBuilder.getJobModelPathPrefix()});
}
@@ -82,11 +100,13 @@ public class ZkControllerImpl implements ZkController {
}
// Only by Leader
- class ZkProcessorChangeHandler implements IZkChildListener {
+ class ZkProcessorChangeHandler implements IZkChildListener {
private final ScheduleAfterDebounceTime debounceTimer;
+
public ZkProcessorChangeHandler(ScheduleAfterDebounceTime debounceTimer) {
this.debounceTimer = debounceTimer;
}
+
/**
* Called when the children of the given path changed.
*
@@ -106,11 +126,14 @@ public class ZkControllerImpl implements ZkController {
class ZkJobModelVersionChangeHandler implements IZkDataListener {
private final ScheduleAfterDebounceTime debounceTimer;
+
public ZkJobModelVersionChangeHandler(ScheduleAfterDebounceTime debounceTimer) {
this.debounceTimer = debounceTimer;
}
+
/**
* called when job model version gets updated
+ *
* @param dataPath
* @param data
* @throws Exception
@@ -123,6 +146,7 @@ public class ZkControllerImpl implements ZkController {
debounceTimer
.scheduleAfterDebounceTime(ScheduleAfterDebounceTime.JOB_MODEL_VERSION_CHANGE, 0, () -> notifyJobModelChange((String) data));
}
+
@Override
public void handleDataDeleted(String dataPath) throws Exception {
throw new SamzaException("version update path has been deleted!.");
@@ -130,7 +154,7 @@ public class ZkControllerImpl implements ZkController {
}
public void shutdown() {
- if(debounceTimer != null)
+ if (debounceTimer != null)
debounceTimer.stopScheduler();
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/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 f661547..e83b16b 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
@@ -1,17 +1,30 @@
+/*
+ * 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.zk;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
import org.apache.samza.SamzaException;
import org.apache.samza.config.Config;
import org.apache.samza.config.JavaSystemConfig;
-import org.apache.samza.config.MapConfig;
import org.apache.samza.coordinator.JobCoordinator;
import org.apache.samza.coordinator.JobModelManager;
import org.apache.samza.coordinator.JobModelManager$;
-import org.apache.samza.job.model.ContainerModel;
import org.apache.samza.job.model.JobModel;
import org.apache.samza.processor.SamzaContainerController;
import org.apache.samza.system.StreamMetadataCache;
@@ -22,7 +35,11 @@ import org.apache.samza.util.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
/**
* JobCoordinator for stand alone processor managed via Zookeeper.
@@ -60,12 +77,11 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
barrier = new ZkBarrierForVersionUpgrade(zkUtils, debounceTimer); //should not have any state in it
-
// TEMP for model generation
//////////////////////////////// NEEDS TO BE REPLACED //////////////////////////////////////
JavaSystemConfig systemConfig = new JavaSystemConfig(this.config);
Map<String, SystemAdmin> systemAdmins = new HashMap<>();
- for (String systemName: systemConfig.getSystemNames()) {
+ for (String systemName : systemConfig.getSystemNames()) {
String systemFactoryClassName = systemConfig.getSystemFactory(systemName);
if (systemFactoryClassName == null) {
log.error(String.format("A stream uses system %s, which is missing from the configuration.", systemName));
@@ -75,11 +91,10 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
systemAdmins.put(systemName, systemFactory.getAdmin(systemName, this.config));
}
- streamMetadataCache = new StreamMetadataCache(Util.<String, SystemAdmin>javaMapAsScalaMap(systemAdmins), 5000, SystemClock
+ streamMetadataCache = new StreamMetadataCache(Util.<String, SystemAdmin>javaMapAsScalaMap(systemAdmins), 5000, SystemClock
.instance());
-
////////////////////////////////////////////////////////////////////////////////////////////
}
@@ -99,8 +114,7 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
}
@Override
- public int getProcessorId()
- {
+ public int getProcessorId() {
return processorId;
}
@@ -125,9 +139,9 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
List<String> currentProcessors = zkUtils.getActiveProcessors();
// get the current version
- String currentJMVersion = zkUtils.getJobModelVersion();
+ String currentJMVersion = zkUtils.getJobModelVersion();
String nextJMVersion;
- if(currentJMVersion == null)
+ if (currentJMVersion == null)
nextJMVersion = "1";
else
nextJMVersion = Integer.toString(Integer.valueOf(currentJMVersion) + 1);
@@ -139,7 +153,7 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
//JobModel jobModel = new JobModel(config, containers);
StringBuilder sb = new StringBuilder();
List<Integer> containerIds = new ArrayList<>();
- for(String processor: currentProcessors){
+ for (String processor : currentProcessors) {
String zkProcessorId = keyBuilder.parseContainerIdFromProcessorId(processor);
sb.append(zkProcessorId).append(",");
containerIds.add(Integer.valueOf(zkProcessorId));
@@ -163,7 +177,7 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
log.info("pid=" + processorId + "published new JobModel ver=" + nextJMVersion);
}
- //////////////////////////////////////////////////////////////////////////////////////////////
+ //////////////////////////////////////////////////////////////////////////////////////////////
@Override
public void onProcessorChange(List<String> processorIds) {
// Reset debounce Timer
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/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 90b0097..02db340 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
@@ -1,3 +1,22 @@
+/*
+ * 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.zk;
import org.apache.samza.config.Config;
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java b/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java
index 42d0c86..efe3349 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkKeyBuilder.java
@@ -1,3 +1,22 @@
+/*
+ * 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.zk;
import org.apache.samza.SamzaException;
@@ -10,10 +29,11 @@ public class ZkKeyBuilder {
public static final String JOBMODEL_VERSION_PATH = "jobModelVersion";
- public ZkKeyBuilder () {
+ public ZkKeyBuilder() {
this("");
}
- public ZkKeyBuilder (String pathPrefix) {
+
+ public ZkKeyBuilder(String pathPrefix) {
this.pathPrefix = pathPrefix;
}
@@ -23,12 +43,12 @@ public class ZkKeyBuilder {
public static String parseIdFromPath(String path) {
if (path != null)
- return path.substring(path.indexOf(PROCESSOR_ID_PREFIX));
+ return path.substring(path.indexOf(PROCESSOR_ID_PREFIX));
return null;
}
public static String parseContainerIdFromProcessorId(String prId) {
- if(prId == null)
+ if (prId == null)
throw new SamzaException("processor id is null");
return prId.substring(prId.indexOf(PROCESSOR_ID_PREFIX) + PROCESSOR_ID_PREFIX.length());
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/zk/ZkLeaderElector.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkLeaderElector.java b/samza-core/src/main/java/org/apache/samza/zk/ZkLeaderElector.java
index 8bffeb6..ecc0f0b 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkLeaderElector.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkLeaderElector.java
@@ -1,12 +1,31 @@
+/*
+ * 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.zk;
-import java.util.Arrays;
import org.I0Itec.zkclient.IZkDataListener;
import org.apache.samza.SamzaException;
import org.apache.samza.coordinator.leaderelection.LeaderElector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Arrays;
import java.util.List;
import java.util.Random;
@@ -23,7 +42,7 @@ public class ZkLeaderElector implements LeaderElector {
private String currentSubscription = null;
private final Random random = new Random();
- public ZkLeaderElector (String processorIdStr, ZkUtils zkUtils, ZkListener zkListener) {
+ public ZkLeaderElector(String processorIdStr, ZkUtils zkUtils, ZkListener zkListener) {
this.processorIdStr = processorIdStr;
this.zkUtils = zkUtils;
this.keyBuilder = this.zkUtils.getKeyBuilder();
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/zk/ZkListener.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkListener.java b/samza-core/src/main/java/org/apache/samza/zk/ZkListener.java
index 4a1c491..a0c69c6 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkListener.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkListener.java
@@ -1,3 +1,22 @@
+/*
+ * 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.zk;
import java.util.List;
http://git-wip-us.apache.org/repos/asf/samza/blob/4918e3ad/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java b/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
index de8c213..05100a5 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
@@ -1,6 +1,24 @@
+/*
+ * 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.zk;
-import java.io.IOException;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.IZkStateListener;
@@ -17,6 +35,7 @@ import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Collections;
@@ -75,8 +94,9 @@ public class ZkUtils {
public ZkKeyBuilder getKeyBuilder() {
return keyBuilder;
}
+
public void makeSurePersistentPathsExists(String[] paths) {
- for(String path: paths) {
+ for (String path : paths) {
if (!zkClient.exists(path)) {
zkClient.createPersistent(path, true);
}
@@ -119,13 +139,14 @@ public class ZkUtils {
throw new SamzaException(e);
}
}
+
public JobModel getJobModel(String jobModelVersion) {
LOG.info("pid=" + processorId + "read the model ver=" + jobModelVersion + " from " + keyBuilder.getJobModelPath(jobModelVersion));
Object data = zkClient.readData(keyBuilder.getJobModelPath(jobModelVersion));
ObjectMapper mmapper = SamzaObjectMapper.getObjectMapper();
JobModel jm;
try {
- jm = mmapper.readValue((String)data, JobModel.class);
+ jm = mmapper.readValue((String) data, JobModel.class);
} catch (IOException e) {
throw new SamzaException("failed to read JobModel from ZK", e);
}
@@ -140,23 +161,24 @@ public class ZkUtils {
public void publishNewJobModelVersion(String oldVersion, String newVersion) {
Stat stat = new Stat();
- String currentVersion = zkClient.<String>readData(keyBuilder.getJobModelVersionPath(), stat);
+ String currentVersion = zkClient.<String>readData(keyBuilder.getJobModelVersionPath(), stat);
LOG.info("pid=" + processorId + " publishing new version: " + newVersion + "; oldVersion = " + oldVersion + "(" + stat.getVersion() + ")");
- if(currentVersion != null && !currentVersion.equals(oldVersion)) {
- throw new SamzaException("Someone change JMVersion while Leader was generating: expected" + oldVersion + ", got " + currentVersion);
+ if (currentVersion != null && !currentVersion.equals(oldVersion)) {
+ throw new SamzaException("Someone change JMVersion while Leader was generating: expected" + oldVersion + ", got " + currentVersion);
}
int dataVersion = stat.getVersion();
stat = zkClient.writeDataReturnStat(keyBuilder.getJobModelVersionPath(), newVersion, dataVersion);
- if(stat.getVersion() != dataVersion + 1)
+ if (stat.getVersion() != dataVersion + 1)
throw new SamzaException("Someone changed data version of the JMVersion while Leader was generating a new one. current= " + dataVersion + ", old version = " + stat.getVersion());
LOG.info("pid=" + processorId +
" published new version: " + newVersion + "; expected dataVersion = " + dataVersion + "(" + stat.getVersion()
- + ")");
+ + ")");
}
/**
* subscribe for changes of JobModel version
+ *
* @param dataListener describe this
*/
public void subscribeToJobModelVersionChange(IZkDataListener dataListener) {
@@ -192,13 +214,13 @@ public class ZkUtils {
}
zkClient.close();
- if(debounceTimer != null)
+ if (debounceTimer != null)
debounceTimer.stopScheduler();
}
public void deleteRoot() {
String rootPath = keyBuilder.getRootPath();
- if(rootPath != null && !rootPath.isEmpty() && zkClient.exists(rootPath)) {
+ if (rootPath != null && !rootPath.isEmpty() && zkClient.exists(rootPath)) {
LOG.info("pid=" + processorId + " Deleteing root: " + rootPath);
zkClient.deleteRecursive(rootPath);
}
@@ -206,6 +228,7 @@ public class ZkUtils {
class ZkStateChangeHandler implements IZkStateListener {
private final ScheduleAfterDebounceTime debounceTimer;
+
public ZkStateChangeHandler(ScheduleAfterDebounceTime debounceTimer) {
this.debounceTimer = debounceTimer;
}
@@ -217,7 +240,8 @@ public class ZkUtils {
* @throws Exception On any error.
*/
@Override
- public void handleStateChanged(Watcher.Event.KeeperState state) throws Exception { }
+ public void handleStateChanged(Watcher.Event.KeeperState state) throws Exception {
+ }
/**
* Called after the zookeeper session has expired and a new session has been created. You would have to re-create
[4/7] samza git commit: added SimpleGroupByContainer
Posted by na...@apache.org.
added SimpleGroupByContainer
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/8cbf0140
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/8cbf0140
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/8cbf0140
Branch: refs/heads/samza-standalone
Commit: 8cbf01407844eab187ea39a222ee7a229b091422
Parents: f15b907
Author: navina <na...@apache.org>
Authored: Fri Dec 23 17:01:04 2016 -0800
Committer: navina <na...@apache.org>
Committed: Fri Dec 23 17:01:04 2016 -0800
----------------------------------------------------------------------
.../task/SimpleGroupByContainerCount.java | 56 ++++++++++++++++++++
.../SimpleGroupByContainerCountFactory.java | 34 ++++++++++++
2 files changed, 90 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/8cbf0140/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java b/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java
new file mode 100644
index 0000000..c9489f7
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCount.java
@@ -0,0 +1,56 @@
+package org.apache.samza.container.grouper.task;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.samza.container.LocalityManager;
+import org.apache.samza.container.TaskName;
+import org.apache.samza.job.model.ContainerModel;
+import org.apache.samza.job.model.TaskModel;
+
+
+public class SimpleGroupByContainerCount implements BalancingTaskNameGrouper {
+ private final int containerCount;
+ public SimpleGroupByContainerCount() {
+ this.containerCount = 1;
+ }
+ public SimpleGroupByContainerCount(int containerCount) {
+ if (containerCount <= 0) throw new IllegalArgumentException("Must have at least one container");
+ this.containerCount = containerCount;
+ }
+
+ @Override
+ public Set<ContainerModel> group(Set<TaskModel> tasks) {
+
+ // Sort tasks by taskName.
+ List<TaskModel> sortedTasks = new ArrayList<>(tasks);
+ Collections.sort(sortedTasks);
+
+ // Map every task to a container in round-robin fashion.
+ Map<TaskName, TaskModel>[] taskGroups = new Map[containerCount];
+ for (int i = 0; i < containerCount; i++) {
+ taskGroups[i] = new HashMap<>();
+ }
+ for (int i = 0; i < sortedTasks.size(); i++) {
+ TaskModel tm = sortedTasks.get(i);
+ taskGroups[i % containerCount].put(tm.getTaskName(), tm);
+ }
+
+ // Convert to a Set of ContainerModel
+ Set<ContainerModel> containerModels = new HashSet<>();
+ for (int i = 0; i < containerCount; i++) {
+ containerModels.add(new ContainerModel(i, taskGroups[i]));
+ }
+
+ return Collections.unmodifiableSet(containerModels);
+ }
+
+ @Override
+ public Set<ContainerModel> balance(Set<TaskModel> tasks, LocalityManager localityManager) {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/8cbf0140/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCountFactory.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCountFactory.java b/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCountFactory.java
new file mode 100644
index 0000000..02918f6
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/container/grouper/task/SimpleGroupByContainerCountFactory.java
@@ -0,0 +1,34 @@
+/*
+ * 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.container.grouper.task;
+
+
+import org.apache.samza.config.Config;
+import org.apache.samza.config.JobConfig;
+
+
+/**
+ * Factory to build the GroupByContainerCount class.
+ */
+public class SimpleGroupByContainerCountFactory implements TaskNameGrouperFactory {
+ @Override
+ public TaskNameGrouper build(Config config) {
+ return new SimpleGroupByContainerCount();
+ }
+}
\ No newline at end of file
[6/7] samza git commit: added configs for EI3 testing, thread naming,
cleanup
Posted by na...@apache.org.
added configs for EI3 testing, thread naming, cleanup
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/0a2b63a4
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/0a2b63a4
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/0a2b63a4
Branch: refs/heads/samza-standalone
Commit: 0a2b63a44ab45a7c3a9b6d8743a5839e7d08cba4
Parents: e37f910
Author: navina <na...@apache.org>
Authored: Fri Dec 23 17:02:45 2016 -0800
Committer: navina <na...@apache.org>
Committed: Fri Dec 23 17:02:45 2016 -0800
----------------------------------------------------------------------
.../apache/samza/processor/SamzaContainerController.java | 10 ++++++++--
.../java/org/apache/samza/processor/StreamProcessor.java | 1 +
.../org/apache/samza/zk/ScheduleAfterDebounceTime.java | 4 +++-
.../main/java/org/apache/samza/zk/ZkJobCoordinator.java | 1 +
samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java | 8 ++++++++
5 files changed, 21 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/0a2b63a4/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java b/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java
index 9352f27..fb227d0 100644
--- a/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java
+++ b/samza-core/src/main/java/org/apache/samza/processor/SamzaContainerController.java
@@ -1,5 +1,6 @@
package org.apache.samza.processor;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.samza.config.ClusterManagerConfig;
import org.apache.samza.config.Config;
import org.apache.samza.config.TaskConfigJava;
@@ -45,8 +46,10 @@ public class SamzaContainerController {
public SamzaContainerController (
Object taskFactory,
long containerShutdownMs,
+ String processorId,
Map<String, MetricsReporter> metricsReporterMap) {
- this.executorService = Executors.newSingleThreadExecutor();
+ this.executorService = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder()
+ .setNameFormat("p" + processorId + "-container-thread-%d").build());
this.taskFactory = taskFactory;
this.metricsReporterMap = metricsReporterMap;
if (containerShutdownMs == -1) {
@@ -73,6 +76,7 @@ public class SamzaContainerController {
if (new ClusterManagerConfig(config).getHostAffinityEnabled()) {
localityManager = SamzaContainer$.MODULE$.getLocalityManager(containerModel.getContainerId(), config);
}
+ log.info("About to create container: " + containerModel.getContainerId());
container = SamzaContainer$.MODULE$.apply(
containerModel.getContainerId(),
containerModel,
@@ -82,6 +86,7 @@ public class SamzaContainerController {
new JmxServer(),
Util.<String, MetricsReporter>javaMapAsScalaMap(metricsReporterMap),
taskFactory);
+ log.info("About to start container: " + containerModel.getContainerId());
containerFuture = executorService.submit(() -> container.run());
}
@@ -109,7 +114,8 @@ public class SamzaContainerController {
container.shutdown();
try {
- containerFuture.get(containerShutdownMs, TimeUnit.MILLISECONDS);
+ if(containerFuture != null)
+ containerFuture.get(containerShutdownMs, TimeUnit.MILLISECONDS);
} catch (InterruptedException | ExecutionException e) {
log.error("Ran into problems while trying to stop the container in the processor!", e);
} catch (TimeoutException e) {
http://git-wip-us.apache.org/repos/asf/samza/blob/0a2b63a4/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 61795e1..596f8f4 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
@@ -121,6 +121,7 @@ public class StreamProcessor {
this.containerController = new SamzaContainerController(
taskFactory,
new TaskConfigJava(updatedConfig).getShutdownMs(),
+ String.valueOf(processorId),
customMetricsReporters);
this.jobCoordinator = Util.
http://git-wip-us.apache.org/repos/asf/samza/blob/0a2b63a4/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java b/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java
index 1854de6..e217dab 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ScheduleAfterDebounceTime.java
@@ -1,5 +1,6 @@
package org.apache.samza.zk;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
@@ -19,7 +20,8 @@ public class ScheduleAfterDebounceTime {
public static final int DEBOUNCE_TIME_MS = 2000;
- private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
+ private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(
+ new ThreadFactoryBuilder().setNameFormat("zk-debounce-thread-%d").setDaemon(true).build());
private final Map<String, ScheduledFuture> futureHandles = new HashMap<>();
public ScheduleAfterDebounceTime () {
http://git-wip-us.apache.org/repos/asf/samza/blob/0a2b63a4/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 560e19b..f661547 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
@@ -95,6 +95,7 @@ public class ZkJobCoordinator implements JobCoordinator, ZkListener {
@Override
public void stop() {
zkController.stop();
+ containerController.shutdown();
}
@Override
http://git-wip-us.apache.org/repos/asf/samza/blob/0a2b63a4/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java b/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
index 0be2b04..de8c213 100644
--- a/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
+++ b/samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java
@@ -185,7 +185,15 @@ public class ZkUtils {
}
public void close() {
+ try {
+ zkConnnection.close();
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
zkClient.close();
+
+ if(debounceTimer != null)
+ debounceTimer.stopScheduler();
}
public void deleteRoot() {