You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ni...@apache.org on 2017/03/18 07:28:47 UTC
samza git commit: SAMZA-1093: instantiating StreamOperatorTask in
SamzaContainer
Repository: samza
Updated Branches:
refs/heads/master eb5d452ae -> 05915bfc8
SAMZA-1093: instantiating StreamOperatorTask in SamzaContainer
SAMZA-1093: Instantiating StreamOperatorTask in SamzaContainer
- minor refactor to get the TaskFactory based on task class configuration
- add instantiation of StreamOperatorTask in SamzaContainer
Author: Yi Pan (Data Infrastructure) <ni...@gmail.com>
Reviewers: Xinyu Liu <xi...@apache.org>, Navina Ramesh <nr...@linkedin.com>, vjagadish1989 <jv...@linkedin.com>, Prateek Maheshwari <pm...@linkedin.com>, Chris Pettitt <cp...@linkedin.com>
Closes #65 from nickpan47/SAMZA-1093 and squashes the following commits:
9daf451 [Yi Pan (Data Infrastructure)] SAMZA-1093: added javadoc for TaskFactoryUtil class
078fee3 [Yi Pan (Data Infrastructure)] Merge branch 'master' into SAMZA-1093
e7be967 [Yi Pan (Data Infrastructure)] SAMZA-1093: addressing review comments
b625beb [Yi Pan (Data Infrastructure)] Merge branch 'master' into SAMZA-1093
df27dc4 [Yi Pan (Data Infrastructure)] Merge branch 'master' into SAMZA-1093
91b1ba8 [Yi Pan (Data Infrastructure)] SAMZA-1093: fix merge errors. Removing 4 renamed old files
12a1529 [Yi Pan (Data Infrastructure)] SAMZA-1093: addressing review feedbacks. Opened SAMZA-1137 for ApplicationRunner initialization in SamzaContainer
ac08d55 [Yi Pan (Data Infrastructure)] WIP: SAMZA-1093: rebase w/ ApplicationRunner patch.
9e1e1f6 [Yi Pan (Data Infrastructure)] SAMZA-1093: addressing review feedbacks.
1c5158d [Yi Pan (Data Infrastructure)] SAMZA-1093: code ready, more unit tests pending
c613334 [Yi Pan (Data Infrastructure)] SAMZA-1093: fix import
6404964 [Yi Pan (Data Infrastructure)] SAMZA-1093: Instantiating StreamOperatorTask in SamzaContainer
b932454 [Yi Pan (Data Infrastructure)] Merge branch 'master' into old-SAMZA-1093
82b839f [Yi Pan (Data Infrastructure)] SAMZA-1093: instantiating StreamOperatorTask in SamzaContainer (WIP)
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/05915bfc
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/05915bfc
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/05915bfc
Branch: refs/heads/master
Commit: 05915bfc8009780d1ef68a13c54bdda53d05b135
Parents: eb5d452
Author: Yi Pan (Data Infrastructure) <ni...@gmail.com>
Authored: Sat Mar 18 00:28:41 2017 -0700
Committer: Yi Pan (Data Infrastructure) <ni...@gmail.com>
Committed: Sat Mar 18 00:28:41 2017 -0700
----------------------------------------------------------------------
.../samza/application/StreamApplication.java | 41 +++
.../samza/operators/StreamGraphBuilder.java | 38 ---
.../apache/samza/runtime/ApplicationRunner.java | 8 +-
.../apache/samza/container/RunLoopFactory.java | 16 +-
.../processor/SamzaContainerController.java | 4 +-
.../samza/runtime/ApplicationRunnerMain.java | 6 +-
.../samza/runtime/LocalApplicationRunner.java | 9 +-
.../samza/runtime/RemoteApplicationRunner.java | 7 +-
.../apache/samza/task/StreamOperatorTask.java | 10 +-
.../org/apache/samza/task/TaskFactoryUtil.java | 175 +++++++++++
.../org/apache/samza/util/ScalaToJavaUtils.java | 41 +++
.../apache/samza/container/SamzaContainer.scala | 66 ++--
.../samza/example/KeyValueStoreExample.java | 4 +-
.../samza/example/NoContextStreamExample.java | 6 +-
.../samza/example/OrderShipmentJoinExample.java | 4 +-
.../samza/example/PageViewCounterExample.java | 4 +-
.../samza/example/RepartitionExample.java | 6 +-
.../apache/samza/example/TestExampleBase.java | 4 +-
.../samza/execution/TestExecutionPlanner.java | 4 +-
.../samza/operators/TestJoinOperator.java | 5 +-
.../runtime/TestAbstractApplicationRunner.java | 4 +-
.../org/apache/samza/task/TestAsyncRunLoop.java | 4 +-
.../apache/samza/task/TestTaskFactoryUtil.java | 298 +++++++++++++++++++
.../testUtils/InvalidStreamApplication.java | 25 ++
.../samza/testUtils/TestAsyncStreamTask.java | 35 +++
.../samza/testUtils/TestStreamApplication.java | 33 ++
.../apache/samza/testUtils/TestStreamTask.java | 34 +++
.../apache/samza/zk/TestZkLeaderElector.java | 2 +-
.../java/org/apache/samza/zk/TestZkUtils.java | 2 +-
.../test/integration/StreamTaskTestUtil.scala | 3 +-
30 files changed, 749 insertions(+), 149 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-api/src/main/java/org/apache/samza/application/StreamApplication.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/application/StreamApplication.java b/samza-api/src/main/java/org/apache/samza/application/StreamApplication.java
new file mode 100644
index 0000000..eeece10
--- /dev/null
+++ b/samza-api/src/main/java/org/apache/samza/application/StreamApplication.java
@@ -0,0 +1,41 @@
+/*
+ * 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.application;
+
+import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.config.Config;
+import org.apache.samza.operators.StreamGraph;
+
+
+/**
+ * This interface defines a template for stream application that user will implement to create operator DAG in {@link StreamGraph}.
+ */
+@InterfaceStability.Unstable
+public interface StreamApplication {
+ static final String APP_CLASS_CONFIG = "app.class";
+
+ /**
+ * Users are required to implement this abstract method to initialize the processing logic of the application, in terms
+ * of a DAG of {@link org.apache.samza.operators.MessageStream}s and operators
+ *
+ * @param graph an empty {@link StreamGraph} object to be initialized
+ * @param config the {@link Config} of the application
+ */
+ void init(StreamGraph graph, Config config);
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-api/src/main/java/org/apache/samza/operators/StreamGraphBuilder.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/operators/StreamGraphBuilder.java b/samza-api/src/main/java/org/apache/samza/operators/StreamGraphBuilder.java
deleted file mode 100644
index b415cf8..0000000
--- a/samza-api/src/main/java/org/apache/samza/operators/StreamGraphBuilder.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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.operators;
-
-import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.config.Config;
-
-
-/**
- * This interface defines a factory class that user will implement to create user-defined operator DAG in a {@link StreamGraph} object.
- */
-@InterfaceStability.Unstable
-public interface StreamGraphBuilder {
- /**
- * Users are required to implement this abstract method to initialize the processing logic of the application, in terms
- * of a DAG of {@link org.apache.samza.operators.MessageStream}s and operators
- *
- * @param graph an empty {@link StreamGraph} object to be initialized
- * @param config the {@link Config} of the application
- */
- void init(StreamGraph graph, Config config);
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-api/src/main/java/org/apache/samza/runtime/ApplicationRunner.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/runtime/ApplicationRunner.java b/samza-api/src/main/java/org/apache/samza/runtime/ApplicationRunner.java
index 553b8d4..6da1242 100644
--- a/samza-api/src/main/java/org/apache/samza/runtime/ApplicationRunner.java
+++ b/samza-api/src/main/java/org/apache/samza/runtime/ApplicationRunner.java
@@ -21,7 +21,7 @@ package org.apache.samza.runtime;
import java.lang.reflect.Constructor;
import org.apache.samza.annotation.InterfaceStability;
import org.apache.samza.config.ConfigException;
-import org.apache.samza.operators.StreamGraphBuilder;
+import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.system.StreamSpec;
@@ -73,11 +73,11 @@ public interface ApplicationRunner {
}
/**
- * Method to be invoked to deploy and run the actual Samza jobs to execute {@link org.apache.samza.operators.StreamGraph}
+ * Method to be invoked to deploy and run the actual Samza jobs to execute {@link StreamApplication}
*
- * @param graphBuilder the user-defined {@link StreamGraphBuilder} object
+ * @param streamApp the user-defined {@link StreamApplication} object
*/
- void run(StreamGraphBuilder graphBuilder);
+ void run(StreamApplication streamApp);
/**
* Constructs a {@link StreamSpec} from the configuration for the specified streamId.
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/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 05e9979..94f02d4 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
@@ -27,12 +27,12 @@ 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;
+import static org.apache.samza.util.ScalaToJavaUtils.defaultValue;
/**
* Factory class to create runloop for a Samza task, based on the type
@@ -112,18 +112,4 @@ public class RunLoopFactory {
}
}
- /**
- * Returns a default value object for scala option.getOrDefault() to use
- * @param value default value
- * @param <T> value type
- * @return object containing default value
- */
- public static <T> AbstractFunction0<T> defaultValue(final T value) {
- return new AbstractFunction0<T>() {
- @Override
- public T apply() {
- return value;
- }
- };
- }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/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 76e2053..81f9d2b 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
@@ -105,7 +105,9 @@ public class SamzaContainerController {
localityManager,
new JmxServer(),
Util.<String, MetricsReporter>javaMapAsScalaMap(metricsReporterMap),
- taskFactory);
+ taskFactory,
+ // TODO: need to use the correct local ApplicationRunner here
+ null);
log.info("About to start container: " + containerModel.getContainerId());
containerFuture = executorService.submit(() -> container.run());
}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/main/java/org/apache/samza/runtime/ApplicationRunnerMain.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/runtime/ApplicationRunnerMain.java b/samza-core/src/main/java/org/apache/samza/runtime/ApplicationRunnerMain.java
index e3f320f..9d124b3 100644
--- a/samza-core/src/main/java/org/apache/samza/runtime/ApplicationRunnerMain.java
+++ b/samza-core/src/main/java/org/apache/samza/runtime/ApplicationRunnerMain.java
@@ -20,9 +20,9 @@
package org.apache.samza.runtime;
import joptsimple.OptionSet;
+import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.job.JobRunner;
-import org.apache.samza.operators.StreamGraphBuilder;
import org.apache.samza.util.CommandLine;
import org.apache.samza.util.Util;
@@ -34,7 +34,7 @@ import org.apache.samza.util.Util;
*/
public class ApplicationRunnerMain {
// TODO: have the app configs consolidated in one place
- private static final String STREAM_APPLICATION_CLASS_CONFIG = "'app.class";
+ private static final String STREAM_APPLICATION_CLASS_CONFIG = "app.class";
public static void main(String[] args) throws Exception {
CommandLine cmdLine = new CommandLine();
@@ -43,7 +43,7 @@ public class ApplicationRunnerMain {
if (config.containsKey(STREAM_APPLICATION_CLASS_CONFIG)) {
ApplicationRunner runner = ApplicationRunner.fromConfig(config);
- StreamGraphBuilder app = (StreamGraphBuilder) Class.forName(config.get(STREAM_APPLICATION_CLASS_CONFIG)).newInstance();
+ StreamApplication app = (StreamApplication) Class.forName(config.get(STREAM_APPLICATION_CLASS_CONFIG)).newInstance();
runner.run(app);
} else {
new JobRunner(Util.rewriteConfig(config)).run(true);
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
index 75cd3d6..d05d9d0 100644
--- a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
+++ b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
@@ -20,7 +20,7 @@
package org.apache.samza.runtime;
import org.apache.samza.config.Config;
-import org.apache.samza.operators.StreamGraphBuilder;
+import org.apache.samza.application.StreamApplication;
/**
@@ -32,14 +32,13 @@ public class LocalApplicationRunner extends AbstractApplicationRunner {
super(config);
}
- @Override
- public void run(StreamGraphBuilder app) {
+ @Override public void run(StreamApplication app) {
// 1. get logic graph for optimization
// StreamGraph logicGraph = this.createGraph(app, config);
// 2. potential optimization....
- // 3. create new instance of StreamGraphBuilder that would generate the optimized graph
+ // 3. create new instance of StreamApplication that would generate the optimized graph
// 4. create all input/output/intermediate topics
// 5. create the configuration for StreamProcessor
- // 6. start the StreamProcessor w/ optimized instance of StreamGraphBuilder
+ // 6. start the StreamProcessor w/ optimized instance of StreamApplication
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/main/java/org/apache/samza/runtime/RemoteApplicationRunner.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/runtime/RemoteApplicationRunner.java b/samza-core/src/main/java/org/apache/samza/runtime/RemoteApplicationRunner.java
index fa13df8..cb61a04 100644
--- a/samza-core/src/main/java/org/apache/samza/runtime/RemoteApplicationRunner.java
+++ b/samza-core/src/main/java/org/apache/samza/runtime/RemoteApplicationRunner.java
@@ -19,12 +19,12 @@
package org.apache.samza.runtime;
+import org.apache.samza.application.StreamApplication;
import org.apache.samza.SamzaException;
import org.apache.samza.execution.ExecutionPlanner;
import org.apache.samza.execution.ProcessorGraph;
import org.apache.samza.job.JobRunner;
import org.apache.samza.operators.StreamGraph;
-import org.apache.samza.operators.StreamGraphBuilder;
import org.apache.samza.config.Config;
import org.apache.samza.operators.StreamGraphImpl;
import org.slf4j.Logger;
@@ -43,11 +43,10 @@ public class RemoteApplicationRunner extends AbstractApplicationRunner {
}
/**
- * Run the {@link StreamGraphBuilder} on the remote cluster
+ * Run the {@link StreamApplication} on the remote cluster
* @param app a StreamApplication
*/
- @Override
- public void run(StreamGraphBuilder app) {
+ @Override public void run(StreamApplication app) {
try {
// 1. build stream graph
StreamGraph streamGraph = new StreamGraphImpl(this, config);
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/main/java/org/apache/samza/task/StreamOperatorTask.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/task/StreamOperatorTask.java b/samza-core/src/main/java/org/apache/samza/task/StreamOperatorTask.java
index fa636ec..9ec8e5a 100644
--- a/samza-core/src/main/java/org/apache/samza/task/StreamOperatorTask.java
+++ b/samza-core/src/main/java/org/apache/samza/task/StreamOperatorTask.java
@@ -23,7 +23,7 @@ import java.util.Map;
import org.apache.samza.config.Config;
import org.apache.samza.operators.ContextManager;
import org.apache.samza.operators.MessageStreamImpl;
-import org.apache.samza.operators.StreamGraphBuilder;
+import org.apache.samza.application.StreamApplication;
import org.apache.samza.operators.StreamGraphImpl;
import org.apache.samza.operators.data.InputMessageEnvelope;
import org.apache.samza.operators.impl.OperatorGraph;
@@ -43,9 +43,9 @@ import org.apache.samza.system.SystemStream;
* This class brings all the operator API implementation components together and feeds the
* {@link InputMessageEnvelope}s into the transformation chains.
* <p>
- * It accepts an instance of the user implemented factory {@link StreamGraphBuilder} as input parameter of the constructor.
+ * It accepts an instance of the user implemented factory {@link StreamApplication} as input parameter of the constructor.
* When its own {@link #init(Config, TaskContext)} method is called during startup, it instantiate a user-defined {@link StreamGraphImpl}
- * from the {@link StreamGraphBuilder}, calls {@link StreamGraphImpl#getContextManager()} to initialize the task-wide context
+ * from the {@link StreamApplication}, calls {@link StreamGraphImpl#getContextManager()} to initialize the task-wide context
* for the graph, and creates a {@link MessageStreamImpl} corresponding to each of its input
* {@link org.apache.samza.system.SystemStreamPartition}s. Each input {@link MessageStreamImpl}
* will be corresponding to either an input stream or intermediate stream in {@link StreamGraphImpl}.
@@ -67,13 +67,13 @@ public final class StreamOperatorTask implements StreamTask, InitableTask, Windo
*/
private final OperatorGraph operatorGraph = new OperatorGraph();
- private final StreamGraphBuilder graphBuilder;
+ private final StreamApplication graphBuilder;
private final ApplicationRunner runner;
private ContextManager contextManager;
- public StreamOperatorTask(StreamGraphBuilder graphBuilder, ApplicationRunner runner) {
+ public StreamOperatorTask(StreamApplication graphBuilder, ApplicationRunner runner) {
this.graphBuilder = graphBuilder;
this.runner = runner;
}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/main/java/org/apache/samza/task/TaskFactoryUtil.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/task/TaskFactoryUtil.java b/samza-core/src/main/java/org/apache/samza/task/TaskFactoryUtil.java
new file mode 100644
index 0000000..e79d278
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/task/TaskFactoryUtil.java
@@ -0,0 +1,175 @@
+/*
+ * 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;
+
+import org.apache.samza.SamzaException;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.ConfigException;
+import org.apache.samza.application.StreamApplication;
+import org.apache.samza.config.TaskConfig;
+import org.apache.samza.runtime.ApplicationRunner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ExecutorService;
+
+import scala.runtime.AbstractFunction0;
+
+/**
+ * This class provides utility functions to load task factory classes based on config, and to wrap {@link StreamTaskFactory} in {@link AsyncStreamTaskFactory}
+ * when running {@link StreamTask}s in multi-thread mode
+ */
+public class TaskFactoryUtil {
+ private static final Logger log = LoggerFactory.getLogger(TaskFactoryUtil.class);
+
+ /**
+ * This method loads a task factory class based on the configuration
+ *
+ * @param config the {@link Config} for this job
+ * @param runner the {@link ApplicationRunner} to run this job
+ * @return a task factory object, either a instance of {@link StreamTaskFactory} or {@link AsyncStreamTaskFactory}
+ */
+ public static Object fromTaskClassConfig(Config config, ApplicationRunner runner) {
+
+ String taskClassName;
+
+ // if there is configuration to set the job w/ a specific type of task, instantiate the corresponding task factory
+ if (isStreamOperatorTask(config)) {
+ taskClassName = StreamOperatorTask.class.getName();
+ } else {
+ taskClassName = new TaskConfig(config).getTaskClass().getOrElse(
+ new AbstractFunction0<String>() {
+ @Override
+ public String apply() {
+ throw new ConfigException("There is no task class defined in the configuration. Failed to create a valid TaskFactory");
+ }
+ });
+ }
+
+ log.info("Got task class name: {}", taskClassName);
+
+ boolean isAsyncTaskClass;
+ try {
+ isAsyncTaskClass = AsyncStreamTask.class.isAssignableFrom(Class.forName(taskClassName));
+ } catch (Throwable t) {
+ throw new ConfigException(String.format("Invalid configuration for AsyncStreamTask class: %s", taskClassName), t);
+ }
+
+ if (isAsyncTaskClass) {
+ return new AsyncStreamTaskFactory() {
+ @Override
+ public AsyncStreamTask createInstance() {
+ try {
+ return (AsyncStreamTask) Class.forName(taskClassName).newInstance();
+ } catch (Throwable t) {
+ log.error("Error loading AsyncStreamTask class: {}. error: {}", taskClassName, t);
+ throw new SamzaException(String.format("Error loading AsyncStreamTask class: %s", taskClassName), t);
+ }
+ }
+ };
+ }
+
+ return new StreamTaskFactory() {
+ @Override
+ public StreamTask createInstance() {
+ try {
+ if (taskClassName.equals(StreamOperatorTask.class.getName())) {
+ return createStreamOperatorTask(config, runner);
+ }
+ return (StreamTask) Class.forName(taskClassName).newInstance();
+ } catch (Throwable t) {
+ log.error("Error loading StreamTask class: {}. error: {}", taskClassName, t);
+ throw new SamzaException(String.format("Error loading StreamTask class: %s", taskClassName), t);
+ }
+ }
+ };
+ }
+
+ /**
+ * Optionally wrap the {@link StreamTaskFactory} in a {@link AsyncStreamTaskFactory}, when running {@link StreamTask}
+ * in multi-thread mode.
+ *
+ * @param factory the task factory instance loaded according to the task class
+ * @param singleThreadMode the flag indicating whether the job is running in single thread mode or not
+ * @param taskThreadPool the thread pool to run the {@link AsyncStreamTaskAdapter} tasks
+ * @return the finalized task factory object
+ */
+ public static Object finalizeTaskFactory(Object factory, boolean singleThreadMode, ExecutorService taskThreadPool) {
+
+ validateFactory(factory);
+
+ boolean isAsyncTaskClass = factory instanceof AsyncStreamTaskFactory;
+ if (isAsyncTaskClass) {
+ log.info("Got an AsyncStreamTask implementation.");
+ }
+
+ if (singleThreadMode && isAsyncTaskClass) {
+ throw new SamzaException("AsyncStreamTask cannot run on single thread mode.");
+ }
+
+ if (!singleThreadMode && !isAsyncTaskClass) {
+ log.info("Converting StreamTask to AsyncStreamTaskAdapter when running StreamTask with multiple threads");
+ return new AsyncStreamTaskFactory() {
+ @Override
+ public AsyncStreamTask createInstance() {
+ return new AsyncStreamTaskAdapter(((StreamTaskFactory) factory).createInstance(), taskThreadPool);
+ }
+ };
+ }
+
+ return factory;
+ }
+
+ private static void validateFactory(Object factory) {
+ if (factory == null) {
+ throw new SamzaException("Either the task class name or the task factory instance is required.");
+ }
+
+ if (!(factory instanceof StreamTaskFactory) && !(factory instanceof AsyncStreamTaskFactory)) {
+ throw new SamzaException(String.format("TaskFactory must be either StreamTaskFactory or AsyncStreamTaskFactory. %s is not supported",
+ factory.getClass()));
+ }
+ }
+
+ private static StreamTask createStreamOperatorTask(Config config, ApplicationRunner runner) throws Exception {
+ Class<?> builderClass = Class.forName(config.get(StreamApplication.APP_CLASS_CONFIG));
+ StreamApplication graphBuilder = (StreamApplication) builderClass.newInstance();
+ return new StreamOperatorTask(graphBuilder, runner);
+ }
+
+ private static boolean isStreamOperatorTask(Config config) {
+ if (config.get(StreamApplication.APP_CLASS_CONFIG) != null && !config.get(StreamApplication.APP_CLASS_CONFIG).isEmpty()) {
+
+ TaskConfig taskConfig = new TaskConfig(config);
+ if (taskConfig.getTaskClass() != null && !taskConfig.getTaskClass().isEmpty()) {
+ throw new ConfigException("High level StreamApplication API cannot be used together with low-level API using task.class.");
+ }
+
+ try {
+ Class<?> builderClass = Class.forName(config.get(StreamApplication.APP_CLASS_CONFIG));
+ return StreamApplication.class.isAssignableFrom(builderClass);
+ } catch (Throwable t) {
+ log.error("Failed to validate StreamApplication class from the config. {}={}",
+ StreamApplication.APP_CLASS_CONFIG, config.get(StreamApplication.APP_CLASS_CONFIG));
+ return false;
+ }
+ }
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/main/java/org/apache/samza/util/ScalaToJavaUtils.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/util/ScalaToJavaUtils.java b/samza-core/src/main/java/org/apache/samza/util/ScalaToJavaUtils.java
new file mode 100644
index 0000000..6c7fc2d
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/util/ScalaToJavaUtils.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.util;
+
+import scala.runtime.AbstractFunction0;
+
+/**
+ * Common utils methods that helps to convert or use Scala objects in Java code
+ */
+public class ScalaToJavaUtils {
+ /**
+ * Returns a default value object for scala option.getOrDefault() to use
+ * @param value default value
+ * @param <T> value type
+ * @return object containing default value
+ */
+ public static <T> AbstractFunction0<T> defaultValue(final T value) {
+ return new AbstractFunction0<T>() {
+ @Override
+ public T apply() {
+ return value;
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/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 b00e10f..77959dd 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
@@ -25,6 +25,7 @@ import java.util
import java.util.concurrent.{CountDownLatch, ExecutorService, Executors, TimeUnit}
import java.lang.Thread.UncaughtExceptionHandler
import java.net.{URL, UnknownHostException}
+
import org.apache.samza.SamzaException
import org.apache.samza.checkpoint.{CheckpointListener, CheckpointManagerFactory, OffsetManager, OffsetManagerMetrics}
import org.apache.samza.config.JobConfig.Config2Job
@@ -48,6 +49,7 @@ import org.apache.samza.metrics.JmxServer
import org.apache.samza.metrics.JvmMetrics
import org.apache.samza.metrics.MetricsRegistryMap
import org.apache.samza.metrics.MetricsReporter
+import org.apache.samza.runtime.ApplicationRunner
import org.apache.samza.serializers.SerdeFactory
import org.apache.samza.serializers.SerdeManager
import org.apache.samza.serializers.model.SamzaObjectMapper
@@ -64,19 +66,12 @@ import org.apache.samza.system.SystemStreamPartition
import org.apache.samza.system.chooser.DefaultChooser
import org.apache.samza.system.chooser.MessageChooserFactory
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.task._
import org.apache.samza.util.HighResolutionClock
import org.apache.samza.util.ExponentialSleepStrategy
import org.apache.samza.util.Logging
import org.apache.samza.util.Throttleable
import org.apache.samza.util.MetricsReporterLoader
-import org.apache.samza.util.ThrottlingExecutor
import org.apache.samza.util.SystemClock
import org.apache.samza.util.Util
import org.apache.samza.util.Util.asScalaClock
@@ -115,6 +110,7 @@ object SamzaContainer extends Logging {
try {
jmxServer = newJmxServer()
val containerModel = jobModel.getContainers.get(containerId.toInt)
+ // TODO: add actual local runner in a container to the parameters
SamzaContainer(
containerId.toInt,
containerModel,
@@ -168,7 +164,9 @@ object SamzaContainer extends Logging {
localityManager: LocalityManager,
jmxServer: JmxServer,
customReporters: Map[String, MetricsReporter] = Map[String, MetricsReporter](),
- taskFactory: Object = null) = {
+ taskFactory: Object = null,
+ // SAMZA-1137: need to instantiate the ApplicationRunner in the container local JVM and pass it in
+ appRunner: ApplicationRunner = null) = {
val containerName = getSamzaContainerName(containerId)
val containerPID = Util.getContainerPID
@@ -432,28 +430,6 @@ 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 cannot run on single thread mode.")
- }
-
val threadPoolSize = config.getThreadPoolSize
info("Got thread pool size: " + threadPoolSize)
@@ -462,6 +438,14 @@ object SamzaContainer extends Logging {
else
null
+ val taskFactoryInstance = Option(taskFactory)
+ .getOrElse(TaskFactoryUtil.fromTaskClassConfig(config, appRunner))
+
+ val finalTaskFactory = TaskFactoryUtil.finalizeTaskFactory(
+ taskFactoryInstance,
+ singleThreadMode,
+ taskThreadPool)
+
// Wire up all task-instance-level (unshared) objects.
val taskNames = containerModel
.getTasks
@@ -483,25 +467,11 @@ object SamzaContainer extends Logging {
val taskName = taskModel.getTaskName
- 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 = finalTaskFactory match {
+ case tf: AsyncStreamTaskFactory => tf.asInstanceOf[AsyncStreamTaskFactory].createInstance()
+ case tf: StreamTaskFactory => tf.asInstanceOf[StreamTaskFactory].createInstance()
}
- val task = if (!singleThreadMode && !isAsyncTask)
- // Wrap the StreamTask into a AsyncStreamTask with the build-in thread pool
- new AsyncStreamTaskAdapter(taskObj.asInstanceOf[StreamTask], taskThreadPool)
- else
- taskObj
-
val taskInstanceMetrics = new TaskInstanceMetrics("TaskName-%s" format taskName)
val collector = new TaskInstanceCollector(producerMultiplexer, taskInstanceMetrics)
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/example/KeyValueStoreExample.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/example/KeyValueStoreExample.java b/samza-core/src/test/java/org/apache/samza/example/KeyValueStoreExample.java
index 7f63f8d..5898f1f 100644
--- a/samza-core/src/test/java/org/apache/samza/example/KeyValueStoreExample.java
+++ b/samza-core/src/test/java/org/apache/samza/example/KeyValueStoreExample.java
@@ -24,7 +24,7 @@ import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.samza.operators.*;
-import org.apache.samza.operators.StreamGraphBuilder;
+import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.operators.data.MessageEnvelope;
import org.apache.samza.operators.functions.FlatMapFunction;
@@ -40,7 +40,7 @@ import org.apache.samza.util.CommandLine;
/**
* Example code using {@link KeyValueStore} to implement event-time window
*/
-public class KeyValueStoreExample implements StreamGraphBuilder {
+public class KeyValueStoreExample implements StreamApplication {
/**
* used by remote application runner to launch the job in remote program. The remote program should follow the similar
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/example/NoContextStreamExample.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/example/NoContextStreamExample.java b/samza-core/src/test/java/org/apache/samza/example/NoContextStreamExample.java
index d6a50bd..516daf5 100644
--- a/samza-core/src/test/java/org/apache/samza/example/NoContextStreamExample.java
+++ b/samza-core/src/test/java/org/apache/samza/example/NoContextStreamExample.java
@@ -19,7 +19,7 @@
package org.apache.samza.example;
import org.apache.samza.operators.*;
-import org.apache.samza.operators.StreamGraphBuilder;
+import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.operators.data.InputMessageEnvelope;
import org.apache.samza.operators.data.JsonIncomingSystemMessageEnvelope;
@@ -38,9 +38,9 @@ import java.util.List;
/**
- * Example {@link StreamGraphBuilder} code to test the API methods
+ * Example {@link StreamApplication} code to test the API methods
*/
-public class NoContextStreamExample implements StreamGraphBuilder {
+public class NoContextStreamExample implements StreamApplication {
StreamSpec input1 = new StreamSpec("inputStreamA", "PageViewEvent", "kafka");
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/example/OrderShipmentJoinExample.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/example/OrderShipmentJoinExample.java b/samza-core/src/test/java/org/apache/samza/example/OrderShipmentJoinExample.java
index 43553f5..a338f6b 100644
--- a/samza-core/src/test/java/org/apache/samza/example/OrderShipmentJoinExample.java
+++ b/samza-core/src/test/java/org/apache/samza/example/OrderShipmentJoinExample.java
@@ -21,8 +21,8 @@ package org.apache.samza.example;
import org.apache.samza.config.Config;
import org.apache.samza.operators.MessageStream;
import org.apache.samza.operators.OutputStream;
+import org.apache.samza.application.StreamApplication;
import org.apache.samza.operators.StreamGraph;
-import org.apache.samza.operators.StreamGraphBuilder;
import org.apache.samza.operators.data.MessageEnvelope;
import org.apache.samza.operators.functions.JoinFunction;
import org.apache.samza.serializers.JsonSerde;
@@ -36,7 +36,7 @@ import java.time.Duration;
/**
* Simple 2-way stream-to-stream join example
*/
-public class OrderShipmentJoinExample implements StreamGraphBuilder {
+public class OrderShipmentJoinExample implements StreamApplication {
/**
* used by remote application runner to launch the job in remote program. The remote program should follow the similar
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/example/PageViewCounterExample.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/example/PageViewCounterExample.java b/samza-core/src/test/java/org/apache/samza/example/PageViewCounterExample.java
index 3e8f856..543716a 100644
--- a/samza-core/src/test/java/org/apache/samza/example/PageViewCounterExample.java
+++ b/samza-core/src/test/java/org/apache/samza/example/PageViewCounterExample.java
@@ -19,7 +19,7 @@
package org.apache.samza.example;
import org.apache.samza.operators.*;
-import org.apache.samza.operators.StreamGraphBuilder;
+import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.operators.data.MessageEnvelope;
import org.apache.samza.operators.triggers.Triggers;
@@ -38,7 +38,7 @@ import java.time.Duration;
/**
* Example code to implement window-based counter
*/
-public class PageViewCounterExample implements StreamGraphBuilder {
+public class PageViewCounterExample implements StreamApplication {
@Override public void init(StreamGraph graph, Config config) {
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/example/RepartitionExample.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/example/RepartitionExample.java b/samza-core/src/test/java/org/apache/samza/example/RepartitionExample.java
index 48e7626..729b26f 100644
--- a/samza-core/src/test/java/org/apache/samza/example/RepartitionExample.java
+++ b/samza-core/src/test/java/org/apache/samza/example/RepartitionExample.java
@@ -19,7 +19,7 @@
package org.apache.samza.example;
import org.apache.samza.operators.*;
-import org.apache.samza.operators.StreamGraphBuilder;
+import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.operators.data.MessageEnvelope;
import org.apache.samza.operators.windows.WindowPane;
@@ -34,9 +34,9 @@ import java.time.Duration;
/**
- * Example {@link StreamGraphBuilder} code to test the API methods with re-partition operator
+ * Example {@link StreamApplication} code to test the API methods with re-partition operator
*/
-public class RepartitionExample implements StreamGraphBuilder {
+public class RepartitionExample implements StreamApplication {
/**
* used by remote application runner to launch the job in remote program. The remote program should follow the similar
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/example/TestExampleBase.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/example/TestExampleBase.java b/samza-core/src/test/java/org/apache/samza/example/TestExampleBase.java
index c4df9d4..dd661a0 100644
--- a/samza-core/src/test/java/org/apache/samza/example/TestExampleBase.java
+++ b/samza-core/src/test/java/org/apache/samza/example/TestExampleBase.java
@@ -18,7 +18,7 @@
*/
package org.apache.samza.example;
-import org.apache.samza.operators.StreamGraphBuilder;
+import org.apache.samza.application.StreamApplication;
import org.apache.samza.system.SystemStream;
import org.apache.samza.system.SystemStreamPartition;
@@ -31,7 +31,7 @@ import java.util.Set;
* Base class for test examples
*
*/
-public abstract class TestExampleBase implements StreamGraphBuilder {
+public abstract class TestExampleBase implements StreamApplication {
protected final Map<SystemStream, Set<SystemStreamPartition>> inputs;
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java b/samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java
index 5b0178d..fa02e04 100644
--- a/samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java
+++ b/samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java
@@ -24,12 +24,12 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.apache.samza.Partition;
+import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.config.JobConfig;
import org.apache.samza.config.MapConfig;
import org.apache.samza.operators.MessageStream;
import org.apache.samza.operators.StreamGraph;
-import org.apache.samza.operators.StreamGraphBuilder;
import org.apache.samza.operators.StreamGraphImpl;
import org.apache.samza.operators.functions.JoinFunction;
import org.apache.samza.operators.functions.SinkFunction;
@@ -201,7 +201,7 @@ public class TestExecutionPlanner {
runner = new AbstractApplicationRunner(config) {
@Override
- public void run(StreamGraphBuilder graphBuilder) {
+ public void run(StreamApplication streamApp) {
}
};
}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java b/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java
index fe2f181..07c6a97 100644
--- a/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java
+++ b/samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java
@@ -20,6 +20,7 @@ package org.apache.samza.operators;
import com.google.common.collect.ImmutableSet;
import org.apache.samza.Partition;
+import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.operators.data.MessageEnvelope;
import org.apache.samza.operators.functions.JoinFunction;
@@ -64,7 +65,7 @@ public class TestJoinOperator {
new SystemStreamPartition("insystem2", "instream2", new Partition(0))));
Config config = mock(Config.class);
- StreamGraphBuilder sgb = new TestStreamGraphBuilder();
+ StreamApplication sgb = new TestStreamApplication();
sot = new StreamOperatorTask(sgb, runner);
sot.init(config, taskContext);
}
@@ -202,7 +203,7 @@ public class TestJoinOperator {
assertTrue(output.isEmpty());
}
- private class TestStreamGraphBuilder implements StreamGraphBuilder {
+ private class TestStreamApplication implements StreamApplication {
StreamSpec inStreamSpec = new StreamSpec("instream", "instream", "insystem");
StreamSpec inStreamSpec2 = new StreamSpec("instream2", "instream2", "insystem2");
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/runtime/TestAbstractApplicationRunner.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/runtime/TestAbstractApplicationRunner.java b/samza-core/src/test/java/org/apache/samza/runtime/TestAbstractApplicationRunner.java
index 631df67..155a47d 100644
--- a/samza-core/src/test/java/org/apache/samza/runtime/TestAbstractApplicationRunner.java
+++ b/samza-core/src/test/java/org/apache/samza/runtime/TestAbstractApplicationRunner.java
@@ -24,7 +24,7 @@ import org.apache.samza.config.Config;
import org.apache.samza.config.JobConfig;
import org.apache.samza.config.MapConfig;
import org.apache.samza.config.StreamConfig;
-import org.apache.samza.operators.StreamGraphBuilder;
+import org.apache.samza.application.StreamApplication;
import org.apache.samza.system.StreamSpec;
import org.junit.Test;
@@ -326,7 +326,7 @@ public class TestAbstractApplicationRunner {
}
@Override
- public void run(StreamGraphBuilder graphBuilder) {
+ public void run(StreamApplication streamApp) {
// do nothing
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/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 e67c3e2..6067701 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
@@ -523,8 +523,8 @@ public class TestAsyncRunLoop {
public void testCommitBehaviourWhenAsyncCommitIsEnabled() throws InterruptedException {
commitRequest = TaskCoordinator.RequestScope.CURRENT_TASK;
maxMessagesInFlight = 2;
- TestTask task0 = new TestTask(true, true, false);
- TestTask task1 = new TestTask(true, false, false);
+ task0 = new TestTask(true, true, false);
+ task1 = new TestTask(true, false, false);
IncomingMessageEnvelope firstMsg = new IncomingMessageEnvelope(ssp0, "0", "key0", "value0");
IncomingMessageEnvelope secondMsg = new IncomingMessageEnvelope(ssp0, "1", "key1", "value1");
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/task/TestTaskFactoryUtil.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/task/TestTaskFactoryUtil.java b/samza-core/src/test/java/org/apache/samza/task/TestTaskFactoryUtil.java
new file mode 100644
index 0000000..27168f3
--- /dev/null
+++ b/samza-core/src/test/java/org/apache/samza/task/TestTaskFactoryUtil.java
@@ -0,0 +1,298 @@
+/*
+ * 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;
+
+import org.apache.samza.SamzaException;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.ConfigException;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.application.StreamApplication;
+import org.apache.samza.runtime.ApplicationRunner;
+import org.apache.samza.testUtils.TestAsyncStreamTask;
+import org.apache.samza.testUtils.TestStreamTask;
+import org.junit.Test;
+
+import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.concurrent.ExecutorService;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Test methods to create {@link StreamTaskFactory} or {@link AsyncStreamTaskFactory} based on task class configuration
+ */
+public class TestTaskFactoryUtil {
+
+ private final ApplicationRunner mockRunner = mock(ApplicationRunner.class);
+
+ @Test
+ public void testStreamTaskClass() {
+ Config config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put("task.class", "org.apache.samza.testUtils.TestStreamTask");
+ }
+ });
+ Object retFactory = TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ assertTrue(retFactory instanceof StreamTaskFactory);
+ assertTrue(((StreamTaskFactory) retFactory).createInstance() instanceof TestStreamTask);
+
+ config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put("task.class", "no.such.class");
+ }
+ });
+ try {
+ TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ fail("Should have failed w/ no.such.class");
+ } catch (ConfigException cfe) {
+ // expected
+ }
+ }
+
+ @Test
+ public void testStreamOperatorTaskClass() {
+ Config config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put(StreamApplication.APP_CLASS_CONFIG, "org.apache.samza.testUtils.TestStreamApplication");
+ }
+ });
+ Object retFactory = TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ assertTrue(retFactory instanceof StreamTaskFactory);
+ assertTrue(((StreamTaskFactory) retFactory).createInstance() instanceof StreamOperatorTask);
+
+ config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put(StreamApplication.APP_CLASS_CONFIG, "org.apache.samza.testUtils.InvalidStreamApplication");
+ }
+ });
+ try {
+ TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ fail("Should have failed w/ no.such.class");
+ } catch (ConfigException ce) {
+ // expected
+ }
+
+ config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put(StreamApplication.APP_CLASS_CONFIG, "no.such.class");
+ }
+ });
+ try {
+ TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ fail("Should have failed w/ no.such.class");
+ } catch (ConfigException ce) {
+ // expected
+ }
+
+ config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put(StreamApplication.APP_CLASS_CONFIG, "");
+ }
+ });
+ try {
+ TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ fail("Should have failed w/ empty class name for StreamApplication");
+ } catch (ConfigException ce) {
+ // expected
+ }
+
+ config = new MapConfig(new HashMap<>());
+ try {
+ TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ fail(String.format("Should have failed w/ non-existing entry for %s", StreamApplication.APP_CLASS_CONFIG));
+ } catch (ConfigException ce) {
+ // expected
+ }
+ }
+
+ @Test
+ public void testStreamOperatorTaskClassWithTaskClass() {
+ Config config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put(StreamApplication.APP_CLASS_CONFIG, "org.apache.samza.testUtils.TestStreamApplication");
+ }
+ });
+ Object retFactory = TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ assertTrue(retFactory instanceof StreamTaskFactory);
+ assertTrue(((StreamTaskFactory) retFactory).createInstance() instanceof StreamOperatorTask);
+
+ config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put("task.class", "org.apache.samza.testUtils.TestAsyncStreamTask");
+ this.put(StreamApplication.APP_CLASS_CONFIG, "org.apache.samza.testUtils.TestStreamApplication");
+ }
+ });
+ try {
+ TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ fail("should have failed with invalid config");
+ } catch (ConfigException ce) {
+ // expected
+ }
+
+ config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put("task.class", "no.such.class");
+ this.put(StreamApplication.APP_CLASS_CONFIG, "org.apache.samza.testUtils.TestStreamApplication");
+ }
+ });
+ try {
+ TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ fail("should have failed with invalid config");
+ } catch (ConfigException ce) {
+ // expected
+ }
+ }
+
+ @Test
+ public void testStreamTaskClassWithInvalidStreamGraphBuilder() {
+
+ Config config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put(StreamApplication.APP_CLASS_CONFIG, "org.apache.samza.testUtils.InvalidStreamApplication");
+ }
+ });
+ try {
+ TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ fail("should have failed with invalid config");
+ } catch (ConfigException ce) {
+ // expected
+ }
+
+ config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put("task.class", "org.apache.samza.testUtils.TestStreamTask");
+ this.put(StreamApplication.APP_CLASS_CONFIG, "");
+ }
+ });
+ Object retFactory = TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ assertTrue(retFactory instanceof StreamTaskFactory);
+ assertTrue(((StreamTaskFactory) retFactory).createInstance() instanceof TestStreamTask);
+
+ config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put("task.class", "");
+ this.put(StreamApplication.APP_CLASS_CONFIG, "org.apache.samza.testUtils.InvalidStreamApplication");
+ }
+ });
+ try {
+ TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ fail("Should have failed w/ no class not found");
+ } catch (ConfigException cne) {
+ // expected
+ }
+ }
+
+ @Test
+ public void testAsyncStreamTask() {
+ Config config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put("task.class", "org.apache.samza.testUtils.TestAsyncStreamTask");
+ }
+ });
+ Object retFactory = TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ assertTrue(retFactory instanceof AsyncStreamTaskFactory);
+ assertTrue(((AsyncStreamTaskFactory) retFactory).createInstance() instanceof TestAsyncStreamTask);
+
+ config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put("task.class", "no.such.class");
+ }
+ });
+ try {
+ TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ fail("Should have failed w/ no.such.class");
+ } catch (ConfigException cfe) {
+ // expected
+ }
+ }
+
+ @Test
+ public void testAsyncStreamTaskWithInvalidStreamGraphBuilder() throws ClassNotFoundException {
+
+ Config config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put(StreamApplication.APP_CLASS_CONFIG, "org.apache.samza.testUtils.InvalidStreamApplication");
+ }
+ });
+ try {
+ TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ fail("Should have failed w/ no.such.class");
+ } catch (ConfigException cfe) {
+ // expected
+ }
+
+ config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put("task.class", "org.apache.samza.testUtils.TestAsyncStreamTask");
+ this.put(StreamApplication.APP_CLASS_CONFIG, "");
+ }
+ });
+ Object retFactory = TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ assertTrue(retFactory instanceof AsyncStreamTaskFactory);
+ assertTrue(((AsyncStreamTaskFactory) retFactory).createInstance() instanceof TestAsyncStreamTask);
+
+ config = new MapConfig(new HashMap<String, String>() {
+ {
+ this.put("task.class", "org.apache.samza.testUtils.TestAsyncStreamTask");
+ this.put(StreamApplication.APP_CLASS_CONFIG, null);
+ }
+ });
+ retFactory = TaskFactoryUtil.fromTaskClassConfig(config, mockRunner);
+ assertTrue(retFactory instanceof AsyncStreamTaskFactory);
+ assertTrue(((AsyncStreamTaskFactory) retFactory).createInstance() instanceof TestAsyncStreamTask);
+ }
+
+ @Test
+ public void testFinalizeTaskFactory() throws NoSuchFieldException, IllegalAccessException {
+ Object mockFactory = mock(Object.class);
+ try {
+ TaskFactoryUtil.finalizeTaskFactory(mockFactory, true, null);
+ fail("Should have failed with validation");
+ } catch (SamzaException se) {
+ // expected
+ }
+ StreamTaskFactory mockStreamFactory = mock(StreamTaskFactory.class);
+ Object retFactory = TaskFactoryUtil.finalizeTaskFactory(mockStreamFactory, true, null);
+ assertEquals(retFactory, mockStreamFactory);
+
+ ExecutorService mockThreadPool = mock(ExecutorService.class);
+ retFactory = TaskFactoryUtil.finalizeTaskFactory(mockStreamFactory, false, mockThreadPool);
+ assertTrue(retFactory instanceof AsyncStreamTaskFactory);
+ assertTrue(((AsyncStreamTaskFactory) retFactory).createInstance() instanceof AsyncStreamTaskAdapter);
+ AsyncStreamTaskAdapter taskAdapter = (AsyncStreamTaskAdapter) ((AsyncStreamTaskFactory) retFactory).createInstance();
+ Field executorSrvFld = AsyncStreamTaskAdapter.class.getDeclaredField("executor");
+ executorSrvFld.setAccessible(true);
+ ExecutorService executor = (ExecutorService) executorSrvFld.get(taskAdapter);
+ assertEquals(executor, mockThreadPool);
+
+ AsyncStreamTaskFactory mockAsyncStreamFactory = mock(AsyncStreamTaskFactory.class);
+ try {
+ TaskFactoryUtil.finalizeTaskFactory(mockAsyncStreamFactory, true, null);
+ fail("Should have failed");
+ } catch (SamzaException se) {
+ // expected
+ }
+
+ retFactory = TaskFactoryUtil.finalizeTaskFactory(mockAsyncStreamFactory, false, null);
+ assertEquals(retFactory, mockAsyncStreamFactory);
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/testUtils/InvalidStreamApplication.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/testUtils/InvalidStreamApplication.java b/samza-core/src/test/java/org/apache/samza/testUtils/InvalidStreamApplication.java
new file mode 100644
index 0000000..5026066
--- /dev/null
+++ b/samza-core/src/test/java/org/apache/samza/testUtils/InvalidStreamApplication.java
@@ -0,0 +1,25 @@
+/*
+ * 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.testUtils;
+
+/**
+ * Test class. Invalid class to implement {@link org.apache.samza.application.StreamApplication}
+ */
+public class InvalidStreamApplication {
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/testUtils/TestAsyncStreamTask.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/testUtils/TestAsyncStreamTask.java b/samza-core/src/test/java/org/apache/samza/testUtils/TestAsyncStreamTask.java
new file mode 100644
index 0000000..81f3fd4
--- /dev/null
+++ b/samza-core/src/test/java/org/apache/samza/testUtils/TestAsyncStreamTask.java
@@ -0,0 +1,35 @@
+/*
+ * 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.testUtils;
+
+import org.apache.samza.system.IncomingMessageEnvelope;
+import org.apache.samza.task.AsyncStreamTask;
+import org.apache.samza.task.MessageCollector;
+import org.apache.samza.task.TaskCallback;
+import org.apache.samza.task.TaskCoordinator;
+
+/**
+ * Test implementation class for {@link AsyncStreamTask}
+ */
+public class TestAsyncStreamTask implements AsyncStreamTask {
+ @Override
+ public void processAsync(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator, TaskCallback callback) {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/testUtils/TestStreamApplication.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/testUtils/TestStreamApplication.java b/samza-core/src/test/java/org/apache/samza/testUtils/TestStreamApplication.java
new file mode 100644
index 0000000..a1cba7d
--- /dev/null
+++ b/samza-core/src/test/java/org/apache/samza/testUtils/TestStreamApplication.java
@@ -0,0 +1,33 @@
+/*
+ * 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.testUtils;
+
+import org.apache.samza.config.Config;
+import org.apache.samza.operators.StreamGraph;
+import org.apache.samza.application.StreamApplication;
+
+/**
+ * Test implementation class for {@link StreamApplication}
+ */
+public class TestStreamApplication implements StreamApplication {
+ @Override
+ public void init(StreamGraph graph, Config config) {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/testUtils/TestStreamTask.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/testUtils/TestStreamTask.java b/samza-core/src/test/java/org/apache/samza/testUtils/TestStreamTask.java
new file mode 100644
index 0000000..ce0980a
--- /dev/null
+++ b/samza-core/src/test/java/org/apache/samza/testUtils/TestStreamTask.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.testUtils;
+
+import org.apache.samza.system.IncomingMessageEnvelope;
+import org.apache.samza.task.MessageCollector;
+import org.apache.samza.task.StreamTask;
+import org.apache.samza.task.TaskCoordinator;
+
+/**
+ * Test implementation class for {@link StreamTask}
+ */
+public class TestStreamTask implements StreamTask {
+ @Override
+ public void process(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator) throws Exception {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/zk/TestZkLeaderElector.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/zk/TestZkLeaderElector.java b/samza-core/src/test/java/org/apache/samza/zk/TestZkLeaderElector.java
index 12fa922..7813da2 100644
--- a/samza-core/src/test/java/org/apache/samza/zk/TestZkLeaderElector.java
+++ b/samza-core/src/test/java/org/apache/samza/zk/TestZkLeaderElector.java
@@ -56,7 +56,7 @@ public class TestZkLeaderElector {
@Before
public void testSetup() {
- testZkConnectionString = "localhost:" + zkServer.getPort();
+ testZkConnectionString = "127.0.0.1:" + zkServer.getPort();
try {
testZkUtils = getZkUtilsWithNewClient();
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-core/src/test/java/org/apache/samza/zk/TestZkUtils.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/zk/TestZkUtils.java b/samza-core/src/test/java/org/apache/samza/zk/TestZkUtils.java
index 913bd49..fba6d0f 100644
--- a/samza-core/src/test/java/org/apache/samza/zk/TestZkUtils.java
+++ b/samza-core/src/test/java/org/apache/samza/zk/TestZkUtils.java
@@ -55,7 +55,7 @@ public class TestZkUtils {
public void testSetup() {
try {
zkClient = new ZkClient(
- new ZkConnection("localhost:" + zkServer.getPort(), SESSION_TIMEOUT_MS),
+ new ZkConnection("127.0.0.1:" + zkServer.getPort(), SESSION_TIMEOUT_MS),
CONNECTION_TIMEOUT_MS);
} catch (Exception e) {
Assert.fail("Client connection setup failed. Aborting tests..");
http://git-wip-us.apache.org/repos/asf/samza/blob/05915bfc/samza-test/src/test/scala/org/apache/samza/test/integration/StreamTaskTestUtil.scala
----------------------------------------------------------------------
diff --git a/samza-test/src/test/scala/org/apache/samza/test/integration/StreamTaskTestUtil.scala b/samza-test/src/test/scala/org/apache/samza/test/integration/StreamTaskTestUtil.scala
index 5d82b92..b803dfe 100644
--- a/samza-test/src/test/scala/org/apache/samza/test/integration/StreamTaskTestUtil.scala
+++ b/samza-test/src/test/scala/org/apache/samza/test/integration/StreamTaskTestUtil.scala
@@ -89,8 +89,6 @@ object StreamTaskTestUtil {
"systems.kafka.samza.offset.default" -> "oldest", // applies to a nonempty topic
"systems.kafka.consumer.auto.offset.reset" -> "smallest", // applies to an empty topic
"systems.kafka.samza.msg.serde" -> "string",
- "systems.kafka.consumer.zookeeper.connect" -> "localhost:2181",
- "systems.kafka.producer.bootstrap.servers" -> "localhost:9092",
// Since using state, need a checkpoint manager
"task.checkpoint.factory" -> "org.apache.samza.checkpoint.kafka.KafkaCheckpointManagerFactory",
"task.checkpoint.system" -> "kafka",
@@ -122,6 +120,7 @@ object StreamTaskTestUtil {
val brokerList = TestUtils.getBrokerListStrFromServers(servers, SecurityProtocol.PLAINTEXT)
brokers = brokerList.split(",").map(p => "localhost" + p).mkString(",")
+ // setup the zookeeper and bootstrap servers for local kafka cluster
jobConfig ++= Map("systems.kafka.consumer.zookeeper.connect" -> zkConnect,
"systems.kafka.producer.bootstrap.servers" -> brokers)