[Streaming] java cross lang streaming graph (#6689)

streaming/java/streaming-runtime/src/main/java/org/ray/streaming/runtime/core/graph/ExecutionEdge.java

@@ -8,7 +8,6 @@ import org.ray.streaming.api.partition.Partition;
  * An edge in the physical execution graph.
  */
 public class ExecutionEdge implements Serializable {
-
   private int srcNodeId;
   private int targetNodeId;
   private Partition partition;

streaming/java/streaming-runtime/src/main/java/org/ray/streaming/runtime/core/graph/ExecutionNode.java

@@ -3,18 +3,17 @@ package org.ray.streaming.runtime.core.graph;
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.List;
+import org.ray.streaming.operator.StreamOperator;
 import org.ray.streaming.plan.VertexType;
-import org.ray.streaming.runtime.core.processor.StreamProcessor;
 
 /**
  * A node in the physical execution graph.
  */
 public class ExecutionNode implements Serializable {
-
   private int nodeId;
   private int parallelism;
   private NodeType nodeType;
-  private StreamProcessor streamProcessor;
+  private StreamOperator streamOperator;
   private List<ExecutionTask> executionTasks;
   private List<ExecutionEdge> inputsEdges;
   private List<ExecutionEdge> outputEdges;
@@ -71,12 +70,12 @@ public class ExecutionNode implements Serializable {
     return inputsEdges;
   }
 
-  public StreamProcessor getStreamProcessor() {
-    return streamProcessor;
+  public StreamOperator getStreamOperator() {
+    return streamOperator;
   }
 
-  public void setStreamProcessor(StreamProcessor streamProcessor) {
-    this.streamProcessor = streamProcessor;
+  public void setStreamOperator(StreamOperator streamOperator) {
+    this.streamOperator = streamOperator;
   }
 
   public NodeType getNodeType() {
@@ -102,7 +101,7 @@ public class ExecutionNode implements Serializable {
     sb.append("nodeId=").append(nodeId);
     sb.append(", parallelism=").append(parallelism);
     sb.append(", nodeType=").append(nodeType);
-    sb.append(", streamProcessor=").append(streamProcessor);
+    sb.append(", streamOperator=").append(streamOperator);
     sb.append('}');
     return sb.toString();
   }

streaming/java/streaming-runtime/src/main/java/org/ray/streaming/runtime/core/graph/ExecutionTask.java

@@ -11,7 +11,6 @@ import org.ray.streaming.runtime.worker.JobWorker;
  * An ExecutionNode has n ExecutionTasks if parallelism is n.
  */
 public class ExecutionTask implements Serializable {
-
   private int taskId;
   private int taskIndex;
   private RayActor<JobWorker> worker;

streaming/java/streaming-runtime/src/main/java/org/ray/streaming/runtime/schedule/TaskAssignerImpl.java

@@ -13,8 +13,6 @@ import org.ray.streaming.runtime.core.graph.ExecutionEdge;
 import org.ray.streaming.runtime.core.graph.ExecutionGraph;
 import org.ray.streaming.runtime.core.graph.ExecutionNode;
 import org.ray.streaming.runtime.core.graph.ExecutionTask;
-import org.ray.streaming.runtime.core.processor.ProcessBuilder;
-import org.ray.streaming.runtime.core.processor.StreamProcessor;
 import org.ray.streaming.runtime.worker.JobWorker;
 
 public class TaskAssignerImpl implements TaskAssigner {
@@ -42,10 +40,8 @@ public class TaskAssignerImpl implements TaskAssigner {
         vertexTasks.add(new ExecutionTask(taskId, taskIndex, workers.get(taskId)));
         taskId++;
       }
-      StreamProcessor streamProcessor = ProcessBuilder
-          .buildProcessor(planVertex.getStreamOperator());
       executionNode.setExecutionTasks(vertexTasks);
-      executionNode.setStreamProcessor(streamProcessor);
+      executionNode.setStreamOperator(planVertex.getStreamOperator());
       idToExecutionNode.put(executionNode.getNodeId(), executionNode);
     }
 

streaming/java/streaming-runtime/src/main/java/org/ray/streaming/runtime/worker/JobWorker.java

@@ -11,6 +11,7 @@ import org.ray.streaming.runtime.core.graph.ExecutionNode;
 import org.ray.streaming.runtime.core.graph.ExecutionNode.NodeType;
 import org.ray.streaming.runtime.core.graph.ExecutionTask;
 import org.ray.streaming.runtime.core.processor.OneInputProcessor;
+import org.ray.streaming.runtime.core.processor.ProcessBuilder;
 import org.ray.streaming.runtime.core.processor.SourceProcessor;
 import org.ray.streaming.runtime.core.processor.StreamProcessor;
 import org.ray.streaming.runtime.transfer.TransferHandler;
@@ -54,7 +55,8 @@ public class JobWorker implements Serializable {
     this.executionNode = executionGraph.getExecutionNodeByTaskId(taskId);
 
     this.nodeType = executionNode.getNodeType();
-    this.streamProcessor = executionNode.getStreamProcessor();
+    this.streamProcessor = ProcessBuilder
+        .buildProcessor(executionNode.getStreamOperator());
     LOGGER.debug("Initializing StreamWorker, taskId: {}, operator: {}.", taskId, streamProcessor);
 
     String channelType = (String) this.config.getOrDefault(

streaming/java/streaming-runtime/src/main/java/org/ray/streaming/runtime/worker/tasks/StreamTask.java

@@ -9,6 +9,7 @@ import org.ray.api.RayActor;
 import org.ray.api.id.ActorId;
 import org.ray.streaming.api.collector.Collector;
 import org.ray.streaming.api.context.RuntimeContext;
+import org.ray.streaming.api.partition.Partition;
 import org.ray.streaming.runtime.core.collector.OutputCollector;
 import org.ray.streaming.runtime.core.graph.ExecutionEdge;
 import org.ray.streaming.runtime.core.graph.ExecutionGraph;
@@ -81,7 +82,8 @@ public abstract class StreamTask implements Runnable {
         DataWriter writer = new DataWriter(channelIDs, toActorIds, queueConf);
         LOG.info("Create DataWriter succeed.");
         writers.put(edge, writer);
-        collectors.add(new OutputCollector(channelIDs, writer, edge.getPartition()));
+        Partition partition = edge.getPartition();
+        collectors.add(new OutputCollector(channelIDs, writer, partition));
       }
     }