Merge remote-tracking branch 'upstream/master'

Author: weisong44

build.gradle

@@ -184,6 +184,7 @@ project(":samza-core_$scalaVersion") {
     compile "org.eclipse.jetty:jetty-webapp:$jettyVersion"
     compile "org.scala-lang:scala-library:$scalaLibVersion"
     compile "org.slf4j:slf4j-api:$slf4jVersion"
+    compile "net.jodah:failsafe:$failsafeVersion"
     testCompile project(":samza-api").sourceSets.test.output
     testCompile "junit:junit:$junitVersion"
     testCompile "org.mockito:mockito-core:$mockitoVersion"

gradle/dependency-versions.gradle

@@ -45,4 +45,5 @@
   yarnVersion = "2.6.1"
   zkClientVersion = "0.8"
   zookeeperVersion = "3.4.6"
+  failsafeVersion = "1.1.0"
 }

samza-core/src/main/java/org/apache/samza/execution/ExecutionPlanner.java

@@ -54,8 +54,8 @@
 
   private final Map<String, JobNode> nodes = new HashMap<>();
   private final Map<String, StreamEdge> edges = new HashMap<>();
-  private final Set<StreamEdge> sources = new HashSet<>();
-  private final Set<StreamEdge> sinks = new HashSet<>();
+  private final Set<StreamEdge> inputStreams = new HashSet<>();
+  private final Set<StreamEdge> outputStreams = new HashSet<>();
   private final Set<StreamEdge> intermediateStreams = new HashSet<>();
   private final Set<TableSpec> tables = new HashSet<>();
   private final Config config;
@@ -115,26 +115,26 @@ public OperatorSpecGraph getSpecGraph() {
 
   /**
    * Add a source stream to a {@link JobNode}
-   * @param input source stream
-   * @param node the job node that consumes from the source
+   * @param streamSpec input stream
+   * @param node the job node that consumes from the streamSpec
    */
-  void addSource(StreamSpec input, JobNode node) {
-    StreamEdge edge = getOrCreateStreamEdge(input);
+  void addInputStream(StreamSpec streamSpec, JobNode node) {
+    StreamEdge edge = getOrCreateStreamEdge(streamSpec);
     edge.addTargetNode(node);
     node.addInEdge(edge);
-    sources.add(edge);
+    inputStreams.add(edge);
   }
 
   /**
-   * Add a sink stream to a {@link JobNode}
-   * @param output sink stream
-   * @param node the job node that outputs to the sink
+   * Add an output stream to a {@link JobNode}
+   * @param streamSpec output stream
+   * @param node the job node that outputs to the output stream
    */
-  void addSink(StreamSpec output, JobNode node) {
-    StreamEdge edge = getOrCreateStreamEdge(output);
+  void addOutputStream(StreamSpec streamSpec, JobNode node) {
+    StreamEdge edge = getOrCreateStreamEdge(streamSpec);
     edge.addSourceNode(node);
     node.addOutEdge(edge);
-    sinks.add(edge);
+    outputStreams.add(edge);
   }
 
   /**
@@ -204,19 +204,19 @@ List<JobNode> getJobNodes() {
   }
 
   /**
-   * Returns the source streams in the graph
+   * Returns the input streams in the graph
    * @return unmodifiable set of {@link StreamEdge}
    */
-  Set<StreamEdge> getSources() {
-    return Collections.unmodifiableSet(sources);
+  Set<StreamEdge> getInputStreams() {
+    return Collections.unmodifiableSet(inputStreams);
   }
 
   /**
-   * Return the sink streams in the graph
+   * Return the output streams in the graph
    * @return unmodifiable set of {@link StreamEdge}
    */
-  Set<StreamEdge> getSinks() {
-    return Collections.unmodifiableSet(sinks);
+  Set<StreamEdge> getOutputStreams() {
+    return Collections.unmodifiableSet(outputStreams);
   }
 
   /**
@@ -236,22 +236,22 @@ Set<StreamEdge> getIntermediateStreamEdges() {
   }
 
   /**
-   * Validate the graph has the correct topology, meaning the sources are coming from external streams,
-   * sinks are going to external streams, and the nodes are connected with intermediate streams.
-   * Also validate all the nodes are reachable from the sources.
+   * Validate the graph has the correct topology, meaning the input streams are coming from external streams,
+   * output streams are going to external streams, and the nodes are connected with intermediate streams.
+   * Also validate all the nodes are reachable from the input streams.
    */
   void validate() {
-    validateSources();
-    validateSinks();
+    validateInputStreams();
+    validateOutputStreams();
     validateInternalStreams();
     validateReachability();
   }
 
   /**
-   * Validate the sources should have indegree being 0 and outdegree greater than 0
+   * Validate the input streams should have indegree being 0 and outdegree greater than 0
    */
-  private void validateSources() {
-    sources.forEach(edge -> {
+  private void validateInputStreams() {
+    inputStreams.forEach(edge -> {
         if (!edge.getSourceNodes().isEmpty()) {
           throw new IllegalArgumentException(
               String.format("Source stream %s should not have producers.", edge.getName()));
@@ -264,10 +264,10 @@ private void validateSources() {
   }
 
   /**
-   * Validate the sinks should have outdegree being 0 and indegree greater than 0
+   * Validate the output streams should have outdegree being 0 and indegree greater than 0
    */
-  private void validateSinks() {
-    sinks.forEach(edge -> {
+  private void validateOutputStreams() {
+    outputStreams.forEach(edge -> {
         if (!edge.getTargetNodes().isEmpty()) {
           throw new IllegalArgumentException(
               String.format("Sink stream %s should not have consumers", edge.getName()));
@@ -284,8 +284,8 @@ private void validateSinks() {
    */
   private void validateInternalStreams() {
     Set<StreamEdge> internalEdges = new HashSet<>(edges.values());
-    internalEdges.removeAll(sources);
-    internalEdges.removeAll(sinks);
+    internalEdges.removeAll(inputStreams);
+    internalEdges.removeAll(outputStreams);
 
     internalEdges.forEach(edge -> {
         if (edge.getSourceNodes().isEmpty() || edge.getTargetNodes().isEmpty()) {
@@ -296,10 +296,10 @@ private void validateInternalStreams() {
   }
 
   /**
-   * Validate all nodes are reachable by sources.
+   * Validate all nodes are reachable by input streams.
    */
   private void validateReachability() {
-    // validate all nodes are reachable from the sources
+    // validate all nodes are reachable from the input streams
     final Set<JobNode> reachable = findReachable();
     if (reachable.size() != nodes.size()) {
       Set<JobNode> unreachable = new HashSet<>(nodes.values());
@@ -317,8 +317,8 @@ Set<JobNode> findReachable() {
     Queue<JobNode> queue = new ArrayDeque<>();
     Set<JobNode> visited = new HashSet<>();
 
-    sources.forEach(source -> {
-        List<JobNode> next = source.getTargetNodes();
+    inputStreams.forEach(input -> {
+        List<JobNode> next = input.getTargetNodes();
         queue.addAll(next);
         visited.addAll(next);
       });
@@ -353,11 +353,11 @@ List<JobNode> topologicalSort() {
     pnodes.forEach(node -> {
         String nid = node.getId();
         //only count the degrees of intermediate streams
-        long degree = node.getInEdges().stream().filter(e -> !sources.contains(e)).count();
+        long degree = node.getInEdges().stream().filter(e -> !inputStreams.contains(e)).count();
         indegree.put(nid, degree);
 
         if (degree == 0L) {
-          // start from the nodes that has no intermediate input streams, so it only consumes from sources
+          // start from the nodes that has no intermediate input streams, so it only consumes from input streams
           q.add(node);
           visited.add(node);
         }
@@ -410,9 +410,9 @@ List<JobNode> topologicalSort() {
           q.add(minNode);
           visited.add(minNode);
         } else {
-          // all the remaining nodes should be reachable from sources
-          // start from sources again to find the next node that hasn't been visited
-          JobNode nextNode = sources.stream().flatMap(source -> source.getTargetNodes().stream())
+          // all the remaining nodes should be reachable from input streams
+          // start from input streams again to find the next node that hasn't been visited
+          JobNode nextNode = inputStreams.stream().flatMap(input -> input.getTargetNodes().stream())
               .filter(node -> !visited.contains(node))
               .findAny().get();
           q.add(nextNode);

samza-core/src/main/java/org/apache/samza/execution/JobGraph.java

@@ -134,8 +134,8 @@ static final class JobGraphJson {
     jobGraphJson.sinkStreams = new HashMap<>();
     jobGraphJson.intermediateStreams = new HashMap<>();
     jobGraphJson.tables = new HashMap<>();
-    jobGraph.getSources().forEach(e -> buildStreamEdgeJson(e, jobGraphJson.sourceStreams));
-    jobGraph.getSinks().forEach(e -> buildStreamEdgeJson(e, jobGraphJson.sinkStreams));
+    jobGraph.getInputStreams().forEach(e -> buildStreamEdgeJson(e, jobGraphJson.sourceStreams));
+    jobGraph.getOutputStreams().forEach(e -> buildStreamEdgeJson(e, jobGraphJson.sinkStreams));
     jobGraph.getIntermediateStreamEdges().forEach(e -> buildStreamEdgeJson(e, jobGraphJson.intermediateStreams));
     jobGraph.getTables().forEach(t -> buildTableJson(t, jobGraphJson.tables));
 

samza-core/src/main/java/org/apache/samza/execution/JobGraphJsonGenerator.java

@@ -37,12 +37,14 @@
 
 /**
  * A {@link TableManager} manages tables within a Samza task. For each table, it maintains
- * the {@link TableSpec} and the {@link TableProvider}. It is used at execution for
- * {@link org.apache.samza.container.TaskInstance} to retrieve table instances for
- * read/write operations.
+ * the {@link TableSpec}, the {@link TableProvider} and the {@link Table} instance.
+ * It is used at execution for {@link org.apache.samza.container.TaskInstance} to retrieve
+ * table instances for read/write operations.
  *
  * A {@link TableManager} is constructed from job configuration, the {@link TableSpec}
- * and {@link TableProvider} are constructed by processing the job configuration.
+ * and {@link TableProvider} are constructed by processing the job configuration
+ * during initialization. The {@link Table} is constructed when {@link #getTable(String)}
+ * is called and cached.
  *
  * After a {@link TableManager} is constructed, local tables are associated with
  * local store instances created during {@link org.apache.samza.container.SamzaContainer}
@@ -51,19 +53,19 @@
  * Method {@link TableManager#getTable(String)} will throw {@link IllegalStateException},
  * if it's called before initialization.
  *
- * For store backed tables, the list of stores must be injected into the constructor.
  */
 public class TableManager {
 
   static public class TableCtx {
     private TableSpec tableSpec;
     private TableProvider tableProvider;
+    private Table table;
   }
 
   private final Logger logger = LoggerFactory.getLogger(TableManager.class.getName());
 
   // tableId -> TableCtx
-  private final Map<String, TableCtx> tables = new HashMap<>();
+  private final Map<String, TableCtx> tableContexts = new HashMap<>();
 
   private boolean initialized;
 
@@ -100,7 +102,7 @@ public TableManager(Config config, Map<String, Serde<Object>> serdes) {
    */
   public void init(SamzaContainerContext containerContext, TaskContext taskContext) {
     Preconditions.checkNotNull(containerContext, "null container context.");
-    tables.values().forEach(ctx -> ctx.tableProvider.init(containerContext, taskContext));
+    tableContexts.values().forEach(ctx -> ctx.tableProvider.init(containerContext, taskContext));
     initialized = true;
   }
 
@@ -109,22 +111,22 @@ public void init(SamzaContainerContext containerContext, TaskContext taskContext
    * @param tableSpec the table spec
    */
   private void addTable(TableSpec tableSpec) {
-    if (tables.containsKey(tableSpec.getId())) {
+    if (tableContexts.containsKey(tableSpec.getId())) {
       throw new SamzaException("Table " + tableSpec.getId() + " already exists");
     }
     TableCtx ctx = new TableCtx();
     TableProviderFactory tableProviderFactory =
         Util.getObj(tableSpec.getTableProviderFactoryClassName(), TableProviderFactory.class);
     ctx.tableProvider = tableProviderFactory.getTableProvider(tableSpec);
     ctx.tableSpec = tableSpec;
-    tables.put(tableSpec.getId(), ctx);
+    tableContexts.put(tableSpec.getId(), ctx);
   }
 
   /**
    * Shutdown the table manager, internally it shuts down all tables
    */
   public void close() {
-    tables.values().forEach(ctx -> ctx.tableProvider.close());
+    tableContexts.values().forEach(ctx -> ctx.tableProvider.close());
   }
 
   /**
@@ -133,10 +135,14 @@ public void close() {
    * @return table instance
    */
   public Table getTable(String tableId) {
-    if (!initialized) {
-      throw new IllegalStateException("TableManager has not been initialized.");
+    Preconditions.checkState(initialized, "TableManager has not been initialized.");
+
+    TableCtx ctx = tableContexts.get(tableId);
+    Preconditions.checkNotNull(ctx, "Unknown tableId " + tableId);
+
+    if (ctx.table == null) {
+      ctx.table = ctx.tableProvider.getTable();
     }
-    Preconditions.checkArgument(tables.containsKey(tableId), "Unknown tableId=" + tableId);
-    return tables.get(tableId).tableProvider.getTable();
+    return ctx.table;
   }
 }

samza-core/src/main/java/org/apache/samza/table/TableManager.java

@@ -43,11 +43,11 @@
  * @param <V> the type of the value in this table
  */
 public class RemoteReadWriteTable<K, V> extends RemoteReadableTable<K, V> implements ReadWriteTable<K, V> {
-  private final TableWriteFunction<K, V> writeFn;
 
   private DefaultTableWriteMetrics writeMetrics;
 
   @VisibleForTesting
+  final TableWriteFunction<K, V> writeFn;
   final TableRateLimiter writeRateLimiter;
 
   public RemoteReadWriteTable(String tableId, TableReadFunction readFn, TableWriteFunction writeFn,

samza-core/src/main/java/org/apache/samza/table/remote/RemoteReadWriteTable.java

@@ -80,10 +80,10 @@ public class RemoteReadableTable<K, V> implements ReadableTable<K, V> {
   protected final ExecutorService callbackExecutor;
   protected final ExecutorService tableExecutor;
 
-  private final TableReadFunction<K, V> readFn;
   private DefaultTableReadMetrics readMetrics;
 
   @VisibleForTesting
+  final TableReadFunction<K, V> readFn;
   final TableRateLimiter<K, V> readRateLimiter;
 
   /**