Merge remote-tracking branch 'upstream/master'

Author: weisong44

.travis.yml

@@ -30,8 +30,8 @@ jdk:
   - oraclejdk8
 
 script:
-  ## travis_wait increases build idle-wait time from 10 minutes to 20 minutes.
-  - travis_wait 20 ./gradlew clean build
+  ## travis_wait increases build idle-wait time from 10 minutes to 30 minutes.
+  - travis_wait 30 ./gradlew clean build
   - type sonar-scanner &>/dev/null; if [ $? -eq 0 ]; then sonar-scanner; else echo "Not running sonar"; fi
 
 before_cache:

samza-api/src/main/java/org/apache/samza/application/ApplicationDescriptor.java

@@ -0,0 +1,80 @@
+/*
+ * 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 java.util.Map;
+import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.config.Config;
+import org.apache.samza.metrics.MetricsReporterFactory;
+import org.apache.samza.operators.ContextManager;
+import org.apache.samza.runtime.ProcessorLifecycleListenerFactory;
+
+
+/**
+ * The interface class to describe the configuration, input and output streams, and processing logic in a {@link SamzaApplication}.
+ * <p>
+ * Sub-classes {@link StreamApplicationDescriptor} and {@link TaskApplicationDescriptor} are specific interfaces for applications
+ * written in high-level {@link StreamApplication} and low-level {@link TaskApplication} APIs, respectively.
+ *
+ * @param <S> sub-class of user application descriptor.
+ */
+@InterfaceStability.Evolving
+public interface ApplicationDescriptor<S extends ApplicationDescriptor> {
+
+  /**
+   * Get the {@link Config} of the application
+   * @return config of the application
+   */
+  Config getConfig();
+
+  /**
+   * Sets the {@link ContextManager} for this application.
+   * <p>
+   * Setting the {@link ContextManager} is optional. The provided {@link ContextManager} can be used to build the shared
+   * context between the operator functions within a task instance
+   *
+   * TODO: this should be replaced by the shared context factory when SAMZA-1714 is fixed.
+
+   * @param contextManager the {@link ContextManager} to use for the application
+   * @return type {@code S} of {@link ApplicationDescriptor} with {@code contextManager} set as its {@link ContextManager}
+   */
+  S withContextManager(ContextManager contextManager);
+
+  /**
+   * Sets the {@link ProcessorLifecycleListenerFactory} for this application.
+   *
+   * <p>Setting a {@link ProcessorLifecycleListenerFactory} is optional to a user application. It allows users to
+   * plug in optional code to be invoked in different stages before/after the main processing logic is started/stopped in
+   * the application.
+   *
+   * @param listenerFactory the user implemented {@link ProcessorLifecycleListenerFactory} that creates lifecycle listener
+   *                        with callback methods before and after the start/stop of each StreamProcessor in the application
+   * @return type {@code S} of {@link ApplicationDescriptor} with {@code listenerFactory} set as its {@link ProcessorLifecycleListenerFactory}
+   */
+  S withProcessorLifecycleListenerFactory(ProcessorLifecycleListenerFactory listenerFactory);
+
+  /**
+   * Sets a set of customized {@link MetricsReporterFactory}s in the application
+   *
+   * @param reporterFactories the map of customized {@link MetricsReporterFactory}s to be used
+   * @return type {@code S} of {@link ApplicationDescriptor} with {@code reporterFactories}
+   */
+  S withMetricsReporterFactories(Map<String, MetricsReporterFactory> reporterFactories);
+
+}

samza-core/src/main/java/org/apache/samza/processor/StreamProcessorLifecycleListener.java renamed to samza-api/src/main/java/org/apache/samza/application/SamzaApplication.java

@@ -16,34 +16,25 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-
-package org.apache.samza.processor;
+package org.apache.samza.application;
 
 import org.apache.samza.annotation.InterfaceStability;
 
 
 /**
- * This class listens to the life cycle events in a {@link StreamProcessor},
+ * The base interface for all user-implemented applications in Samza.
+ * <p>
+ * The main processing logic of the user application should be implemented in {@link SamzaApplication#describe(ApplicationDescriptor)}
+ * method. Sub-classes {@link StreamApplication} and {@link TaskApplication} are specific interfaces for applications
+ * written in high-level DAG and low-level task APIs, respectively.
  */
 @InterfaceStability.Evolving
-public interface StreamProcessorLifecycleListener {
-  /**
-   * Callback when the {@link StreamProcessor} is started
-   * This callback is invoked only once when {@link org.apache.samza.container.SamzaContainer} starts for the first time
-   * in the {@link StreamProcessor}. When there is a re-balance of tasks/partitions among the processors, the container
-   * may temporarily be "paused" and re-started again. For such re-starts, this callback is NOT invoked.
-   */
-  void onStart();
+public interface SamzaApplication<S extends ApplicationDescriptor> {
 
   /**
-   * Callback when the {@link StreamProcessor} is shut down.
+   * Describes the user processing logic via {@link ApplicationDescriptor}
+   *
+   * @param appDesc the {@link ApplicationDescriptor} object to describe user application logic
    */
-  void onShutdown();
-
-  /**
-   * Callback when the {@link StreamProcessor} fails
-   * @param t Cause of the failure
-   */
-  void onFailure(Throwable t);
-
+  void describe(S appDesc);
 }

samza-api/src/main/java/org/apache/samza/application/StreamApplication.java

@@ -19,30 +19,27 @@
 package org.apache.samza.application;
 
 import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.config.Config;
-import org.apache.samza.operators.ContextManager;
-import org.apache.samza.operators.MessageStream;
-import org.apache.samza.operators.OutputStream;
-import org.apache.samza.operators.StreamGraph;
-import org.apache.samza.operators.functions.InitableFunction;
-import org.apache.samza.task.StreamTask;
-import org.apache.samza.task.TaskContext;
 
 /**
- * Describes and initializes the transforms for processing message streams and generating results.
+ * Describes and initializes the transforms for processing message streams and generating results in high-level API. 
  * <p>
  * The following example removes page views older than 1 hour from the input stream:
  * <pre>{@code
- * public class PageViewCounter implements StreamApplication {
- *   public void init(StreamGraph graph, Config config) {
- *     MessageStream<PageViewEvent> pageViewEvents =
- *       graph.getInputStream("pageViewEvents", (k, m) -> (PageViewEvent) m);
- *     OutputStream<String, PageViewEvent, PageViewEvent> recentPageViewEvents =
- *       graph.getOutputStream("recentPageViewEvents", m -> m.memberId, m -> m);
+ * public class PageViewFilter implements StreamApplication {
+ *   public void describe(StreamAppDescriptor appDesc) {
+ *     KafkaSystemDescriptor trackingSystem = new KafkaSystemDescriptor("tracking");
+ *     KafkaInputDescriptor<PageViewEvent> inputStreamDescriptor =
+ *         trackingSystem.getInputDescriptor("pageViewEvent", new JsonSerdeV2<>(PageViewEvent.class));
+ *
+ *     KafkaOutputDescriptor<PageViewEvent>> outputStreamDescriptor =
+ *         trackingSystem.getOutputDescriptor("recentPageViewEvent", new JsonSerdeV2<>(PageViewEvent.class)));
+ *
+ *     MessageStream<PageViewEvent> pageViewEvents = appDesc.getInputStream(inputStreamDescriptor);
+ *     OutputStream<PageViewEvent> recentPageViewEvents = appDesc.getOutputStream(outputStreamDescriptor);
  *
  *     pageViewEvents
  *       .filter(m -> m.getCreationTime() > System.currentTimeMillis() - Duration.ofHours(1).toMillis())
- *       .sendTo(filteredPageViewEvents);
+ *       .sendTo(recentPageViewEvents);
  *   }
  * }
  * }</pre>
@@ -52,46 +49,28 @@
  *   public static void main(String[] args) {
  *     CommandLine cmdLine = new CommandLine();
  *     Config config = cmdLine.loadConfig(cmdLine.parser().parse(args));
- *     PageViewCounter app = new PageViewCounter();
- *     LocalApplicationRunner runner = new LocalApplicationRunner(config);
- *     runner.run(app);
+ *     PageViewFilter app = new PageViewFilter();
+ *     ApplicationRunner runner = ApplicationRunners.getApplicationRunner(app, config);
+ *     runner.run();
  *     runner.waitForFinish();
  *   }
  * }</pre>
  *
  * <p>
- * Implementation Notes: Currently StreamApplications are wrapped in a {@link StreamTask} during execution.
- * A new StreamApplication instance will be created and initialized with a user-defined {@link StreamGraph}
- * when planning the execution. The {@link StreamGraph} and the functions implemented for transforms are required to
- * be serializable. The execution planner will generate a serialized DAG which will be deserialized in each {@link StreamTask}
- * instance used for processing incoming messages. Execution is synchronous and thread-safe within each {@link StreamTask}.
+ * Implementation Notes: Currently {@link StreamApplication}s are wrapped in a {@link org.apache.samza.task.StreamTask}
+ * during execution. The execution planner will generate a serialized DAG which will be deserialized in each
+ * {@link org.apache.samza.task.StreamTask} instance used for processing incoming messages. Execution is synchronous
+ * and thread-safe within each {@link org.apache.samza.task.StreamTask}.
  *
  * <p>
+ * A {@link StreamApplication} implementation must have a proper fully-qualified class name and a default constructor
+ * with no parameters to ensure successful instantiation in both local and remote environments.
  * Functions implemented for transforms in StreamApplications ({@link org.apache.samza.operators.functions.MapFunction},
  * {@link org.apache.samza.operators.functions.FilterFunction} for e.g.) are initable and closable. They are initialized
- * before messages are delivered to them and closed after their execution when the {@link StreamTask} instance is closed.
- * See {@link InitableFunction} and {@link org.apache.samza.operators.functions.ClosableFunction}.
+ * before messages are delivered to them and closed after their execution when the {@link org.apache.samza.task.StreamTask}
+ * instance is closed. See {@link org.apache.samza.operators.functions.InitableFunction} and {@link org.apache.samza.operators.functions.ClosableFunction}.
+ * Function implementations are required to be {@link java.io.Serializable}.
  */
-@InterfaceStability.Unstable
-public interface StreamApplication {
-
-  /**
-   * Describes and initializes the transforms for processing message streams and generating results.
-   * <p>
-   * The {@link StreamGraph} provides access to input and output streams. Input {@link MessageStream}s can be
-   * transformed into other {@link MessageStream}s or sent to an {@link OutputStream} using the {@link MessageStream}
-   * operators.
-   * <p>
-   * Most operators accept custom functions for doing the transformations. These functions are {@link InitableFunction}s
-   * and are provided the {@link Config} and {@link TaskContext} during their own initialization. The config and the
-   * context can be used, for example, to create custom metrics or access durable state stores.
-   * <p>
-   * A shared context between {@link InitableFunction}s for different operators within a task instance can be set
-   * up by providing a {@link ContextManager} using {@link StreamGraph#withContextManager}.
-   *
-   * @param graph the {@link StreamGraph} to get input/output streams from
-   * @param config the configuration for the application
-   */
-  void init(StreamGraph graph, Config config);
-
+@InterfaceStability.Evolving
+public interface StreamApplication extends SamzaApplication<StreamApplicationDescriptor> {
 }

samza-api/src/main/java/org/apache/samza/operators/StreamGraph.java renamed to samza-api/src/main/java/org/apache/samza/application/StreamApplicationDescriptor.java

@@ -16,36 +16,41 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators;
+package org.apache.samza.application;
 
 import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.operators.KV;
+import org.apache.samza.operators.MessageStream;
+import org.apache.samza.operators.OutputStream;
+import org.apache.samza.operators.TableDescriptor;
 import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
 import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
 import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
 import org.apache.samza.table.Table;
 
 
 /**
- * Provides access to {@link MessageStream}s and {@link OutputStream}s used to describe application logic.
+ * The interface class to describe a {@link SamzaApplication} in high-level API in Samza.
  */
-@InterfaceStability.Unstable
-public interface StreamGraph {
+@InterfaceStability.Evolving
+public interface StreamApplicationDescriptor extends ApplicationDescriptor<StreamApplicationDescriptor> {
 
   /**
    * Sets the default SystemDescriptor to use for intermediate streams. This is equivalent to setting
    * {@code job.default.system} and its properties in configuration.
    * <p>
-   * If the default system descriptor is set, it must be set <b>before</b> creating any intermediate streams.
+   * If the default system descriptor is set, it must be set <b>before</b> creating any input/output/intermediate streams.
    * <p>
-   * If the intermediate stream is created with a stream-level Serde, they will be used, else the serde specified
+   * If an input/output stream is created with a stream-level Serde, they will be used, else the serde specified
    * for the {@code job.default.system} in configuration will be used.
    * <p>
    * Providing an incompatible message type for the intermediate streams that use the default serde will result in
    * {@link ClassCastException}s at runtime.
    *
    * @param defaultSystemDescriptor the default system descriptor to use
+   * @return type {@code S} of {@link ApplicationDescriptor} with {@code defaultSystemDescriptor} set as its default system
    */
-  void setDefaultSystem(SystemDescriptor<?> defaultSystemDescriptor);
+  StreamApplicationDescriptor withDefaultSystem(SystemDescriptor<?> defaultSystemDescriptor);
 
   /**
    * Gets the input {@link MessageStream} corresponding to the {@code inputDescriptor}.
@@ -105,16 +110,4 @@ public interface StreamGraph {
    * @throws IllegalStateException when invoked multiple times with the same {@link TableDescriptor}
    */
   <K, V> Table<KV<K, V>> getTable(TableDescriptor<K, V, ?> tableDescriptor);
-
-  /**
-   * Sets the {@link ContextManager} for this {@link StreamGraph}.
-   * <p>
-   * The provided {@link ContextManager} can be used to setup shared context between the operator functions
-   * within a task instance
-   *
-   * @param contextManager the {@link ContextManager} to use for the {@link StreamGraph}
-   * @return the {@link StreamGraph} with {@code contextManager} set as its {@link ContextManager}
-   */
-  StreamGraph withContextManager(ContextManager contextManager);
-
-}
+}

samza-api/src/main/java/org/apache/samza/application/TaskApplication.java

@@ -0,0 +1,86 @@
+/*
+ * 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;
+
+
+/**
+ * Describes and initializes the transforms for processing message streams and generating results in low-level API. Your
+ * application is expected to implement this interface.
+ * <p>
+ * The following example removes page views older than 1 hour from the input stream:
+ * <pre>{@code
+ * public class PageViewFilter implements TaskApplication {
+ *   public void describe(TaskAppDescriptor appDesc) {
+ *     KafkaSystemDescriptor trackingSystem = new KafkaSystemDescriptor(PageViewTask.SYSTEM);
+ *     KafkaInputDescriptor<PageViewEvent> inputStreamDescriptor =
+ *         trackingSystem.getInputDescriptor(PageViewTask.TASK_INPUT, new JsonSerdeV2<>(PageViewEvent.class));
+ *
+ *     KafkaOutputDescriptor<PageViewEvent>> outputStreamDescriptor =
+ *         trackingSystem.getOutputDescriptor(PageViewTask.TASK_OUTPUT, new JsonSerdeV2<>(PageViewEvent.class)));
+ *
+ *     appDesc.addInputStream(inputStreamDescriptor);
+ *     appDesc.addOutputStream(outputStreamDescriptor);
+ *     appDesc.setTaskFactory((StreamTaskFactory) () -> new PageViewTask());
+ *   }
+ * }
+ *
+ * public class PageViewTask implements StreamTask {
+ *   final static String TASK_INPUT = "pageViewEvents";
+ *   final static String TASK_OUTPUT = "recentPageViewEvents";
+ *   final static String SYSTEM = "kafka";
+ *
+ *   public void process(IncomingMessageEnvelope message, MessageCollector collector,
+ *       TaskCoordinator coordinator) {
+ *     PageViewEvent m = (PageViewEvent) message.getValue();
+ *     if (m.getCreationTime() > System.currentTimeMillis() - Duration.ofHours(1).toMillis()) {
+ *       collector.send(new OutgoingMessageEnvelope(new SystemStream(SYSTEM, TASK_OUTPUT),
+ *           message.getKey(), message.getKey(), m));
+ *     }
+ *   }
+ * }
+ * }</pre>
+ *
+ *<p>
+ * The example above can be run using an ApplicationRunner:
+ * <pre>{@code
+ *   public static void main(String[] args) {
+ *     CommandLine cmdLine = new CommandLine();
+ *     Config config = cmdLine.loadConfig(cmdLine.parser().parse(args));
+ *     PageViewFilter app = new PageViewFilter();
+ *     ApplicationRunner runner = ApplicationRunners.getApplicationRunner(app, config);
+ *     runner.run();
+ *     runner.waitForFinish();
+ *   }
+ * }</pre>
+ *
+ * <p>
+ * Implementation Notes: {@link TaskApplication} allow users to instantiate {@link org.apache.samza.task.StreamTask} or
+ * {@link org.apache.samza.task.AsyncStreamTask} when describing the processing logic. A new {@link TaskApplicationDescriptor }
+ * instance will be created and described by the user-defined {@link TaskApplication} when planning the execution.
+ * {@link org.apache.samza.task.TaskFactory} is required to be serializable.
+ *
+ * <p>
+ * The user-implemented {@link TaskApplication} class must be a class with proper fully-qualified class name and
+ * a default constructor with no parameters to ensure successful instantiation in both local and remote environments.
+ */
+@InterfaceStability.Evolving
+public interface TaskApplication extends SamzaApplication<TaskApplicationDescriptor> {
+}