[Streaming] Implement streaming job-worker. (#8780)

2026-07-10 15:19:12 +08:00 · 2020-06-10 14:13:55 +08:00
parent 04cffb7e65
commit ec5ecb661f
56 changed files with 1078 additions and 1213 deletions
@@ -0,0 +1,55 @@
+package io.ray.streaming.runtime.client;
+
+import io.ray.api.ActorHandle;
+import io.ray.api.ObjectRef;
+import io.ray.api.Ray;
+import io.ray.api.options.ActorCreationOptions;
+import io.ray.streaming.client.JobClient;
+import io.ray.streaming.jobgraph.JobGraph;
+import io.ray.streaming.runtime.config.global.CommonConfig;
+import io.ray.streaming.runtime.master.JobMaster;
+import java.util.HashMap;
+import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Job client: to submit job from api to runtime.
+ */
+public class JobClientImpl implements JobClient {
+
+  public static final Logger LOG = LoggerFactory.getLogger(JobClientImpl.class);
+
+  private ActorHandle<JobMaster> jobMasterActor;
+
+  @Override
+  public void submit(JobGraph jobGraph, Map<String, String> jobConfig) {
+    LOG.info("Submitting job [{}] with job graph [{}] and job config [{}].",
+        jobGraph.getJobName(), jobGraph, jobConfig);
+    Map<String, Double> resources = new HashMap<>();
+    ActorCreationOptions options = new ActorCreationOptions.Builder()
+        .setResources(resources)
+        .setMaxRestarts(-1)
+        .createActorCreationOptions();
+
+    // set job name and id at start
+    jobConfig.put(CommonConfig.JOB_ID, Ray.getRuntimeContext().getCurrentJobId().toString());
+    jobConfig.put(CommonConfig.JOB_NAME, jobGraph.getJobName());
+
+    jobGraph.getJobConfig().putAll(jobConfig);
+
+    // create job master actor
+    this.jobMasterActor = Ray.createActor(JobMaster::new, jobConfig, options);
+
+    try {
+      ObjectRef<Boolean> submitResult = jobMasterActor.call(JobMaster::submitJob,
+          jobMasterActor, jobGraph);
+
+      if (submitResult.get()) {
+        LOG.info("Finish submitting job: {}.", jobGraph.getJobName());
+      }
+    } catch (Exception e) {
+      LOG.error("Failed to submit job: {}.", jobGraph.getJobName(), e);
+    }
+  }
+}
@@ -2,6 +2,7 @@ package io.ray.streaming.runtime.config;

 import com.google.common.base.Preconditions;
 import io.ray.streaming.runtime.config.global.CommonConfig;
+import io.ray.streaming.runtime.config.global.TransferConfig;
 import java.io.Serializable;
 import java.lang.reflect.Method;
 import java.util.HashMap;
@@ -20,6 +21,7 @@ public class StreamingGlobalConfig implements Serializable {
  private static final Logger LOG = LoggerFactory.getLogger(StreamingGlobalConfig.class);

  public final CommonConfig commonConfig;
+  public final TransferConfig transferConfig;

  public final Map<String, String> configMap;

@@ -27,6 +29,7 @@ public class StreamingGlobalConfig implements Serializable {
    configMap = new HashMap<>(conf);

    commonConfig = ConfigFactory.create(CommonConfig.class, conf);
+    transferConfig = ConfigFactory.create(TransferConfig.class, conf);
    globalConfig2Map();
  }

@@ -38,6 +41,7 @@ public class StreamingGlobalConfig implements Serializable {
  private void globalConfig2Map() {
    try {
      configMap.putAll(config2Map(this.commonConfig));
+      configMap.putAll(config2Map(this.transferConfig));
    } catch (Exception e) {
      LOG.error("Couldn't convert global config to a map.", e);
    }
@@ -1,6 +1,9 @@
 package io.ray.streaming.runtime.config;

+import io.ray.streaming.runtime.config.worker.WorkerInternalConfig;
+import java.util.HashMap;
 import java.util.Map;
+import org.aeonbits.owner.ConfigFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

@@ -11,7 +14,24 @@ public class StreamingWorkerConfig extends StreamingGlobalConfig {

  private static final Logger LOG = LoggerFactory.getLogger(StreamingWorkerConfig.class);

+  public WorkerInternalConfig workerInternalConfig;
+
  public StreamingWorkerConfig(final Map<String, String> conf) {
    super(conf);
+    workerInternalConfig = ConfigFactory.create(WorkerInternalConfig.class, conf);
+
+    configMap.putAll(workerConfig2Map());
  }
+
+  public Map<String, String> workerConfig2Map() {
+    Map<String, String> result = new HashMap<>();
+    try {
+      result.putAll(config2Map(this.workerInternalConfig));
+    } catch (Exception e) {
+      LOG.error("Worker config to map occur error.", e);
+      return null;
+    }
+    return result;
+  }
+
 }
@@ -0,0 +1,66 @@
+package io.ray.streaming.runtime.config.global;
+
+import io.ray.streaming.runtime.config.Config;
+import io.ray.streaming.runtime.config.types.TransferChannelType;
+
+/**
+ * Job data transfer config.
+ */
+public interface TransferConfig extends Config {
+
+  /**
+   * Data transfer channel type, support memory queue and native queue.
+   */
+  @DefaultValue(value = "NATIVE_CHANNEL")
+  @Key(value = io.ray.streaming.util.Config.CHANNEL_TYPE)
+  TransferChannelType channelType();
+
+  /**
+   * Queue size.
+   */
+  @DefaultValue(value = "100000000")
+  @Key(value = io.ray.streaming.util.Config.CHANNEL_SIZE)
+  long channelSize();
+
+  /**
+   * DataRead read timeout.
+   */
+  @DefaultValue(value = "false")
+  @Key(value = io.ray.streaming.util.Config.IS_RECREATE)
+  boolean readerIsRecreate();
+
+  /**
+   * Return from DataReader.getBundle if only empty message read in this interval.
+   */
+  @DefaultValue(value = "-1")
+  @Key(value = io.ray.streaming.util.Config.TIMER_INTERVAL_MS)
+  long readerTimerIntervalMs();
+
+  /**
+   * Ring capacity.
+   */
+  @DefaultValue(value = "-1")
+  @Key(value = io.ray.streaming.util.Config.STREAMING_RING_BUFFER_CAPACITY)
+  int ringBufferCapacity();
+
+  /**
+   * Write an empty message if there is no data to be written in this interval.
+   */
+  @DefaultValue(value = "-1")
+  @Key(value = io.ray.streaming.util.Config.STREAMING_EMPTY_MESSAGE_INTERVAL)
+  int emptyMsgInterval();
+
+  // Flow control
+
+  @DefaultValue(value = "-1")
+  @Key(value = io.ray.streaming.util.Config.FLOW_CONTROL_TYPE)
+  int flowControlType();
+
+  @DefaultValue(value = "-1")
+  @Key(value = io.ray.streaming.util.Config.WRITER_CONSUMED_STEP)
+  int writerConsumedStep();
+
+  @DefaultValue(value = "-1")
+  @Key(value = io.ray.streaming.util.Config.READER_CONSUMED_STEP)
+  int readerConsumedStep();
+}
@@ -54,14 +54,14 @@ public interface ResourceConfig extends Config {
  /**
   * Whether to enable CPU limit in resource control.
   */
-  @DefaultValue(value = "true")
+  @DefaultValue(value = "false")
  @Key(value = TASK_RESOURCE_CPU_LIMIT_ENABLE)
  boolean isTaskCpuResourceLimit();

  /**
   * Whether to enable memory limit in resource control.
   */
-  @DefaultValue(value = "true")
+  @DefaultValue(value = "false")
  @Key(value = TASK_RESOURCE_MEM_LIMIT_ENABLE)
  boolean isTaskMemResourceLimit();

@@ -0,0 +1,29 @@
+package io.ray.streaming.runtime.config.types;
+
+/**
+ * Data transfer channel type.
+ */
+public enum TransferChannelType {
+
+  /**
+   * Memory queue.
+   */
+  MEMORY_CHANNEL("memory_channel", 0),
+
+  /**
+   * Native queue.
+   */
+  NATIVE_CHANNEL("native_channel", 1);
+
+  private String value;
+  private int index;
+
+  TransferChannelType(String value, int index) {
+    this.value = value;
+    this.index = index;
+  }
+
+  public String getValue() {
+    return value;
+  }
+}
@@ -0,0 +1,27 @@
+package io.ray.streaming.runtime.config.worker;
+
+import io.ray.streaming.runtime.config.Config;
+import org.aeonbits.owner.Mutable;
+
+/**
+ * This worker config is used by JobMaster to define the internal configuration of JobWorker.
+ */
+public interface WorkerInternalConfig extends Config, Mutable {
+
+  String WORKER_NAME_INTERNAL = io.ray.streaming.util.Config.STREAMING_WORKER_NAME;
+  String OP_NAME_INTERNAL = io.ray.streaming.util.Config.STREAMING_OP_NAME;
+
+  /**
+   * The name of the worker inside the system.
+   */
+  @DefaultValue(value = "default-worker-name")
+  @Key(value = WORKER_NAME_INTERNAL)
+  String workerName();
+
+  /**
+   * Operator name corresponding to worker.
+   */
+  @DefaultValue(value = "default-worker-op-name")
+  @Key(value = OP_NAME_INTERNAL)
+  String workerOperatorName();
+}
@@ -1,47 +0,0 @@
-package io.ray.streaming.runtime.core.graph;
-
-import io.ray.streaming.api.partition.Partition;
-import java.io.Serializable;
-
-/**
- * An edge in the physical execution graph.
- */
-public class ExecutionEdge implements Serializable {
-  private int srcNodeId;
-  private int targetNodeId;
-  private Partition partition;
-
-  public ExecutionEdge(int srcNodeId, int targetNodeId, Partition partition) {
-    this.srcNodeId = srcNodeId;
-    this.targetNodeId = targetNodeId;
-    this.partition = partition;
-  }
-
-  public int getSrcNodeId() {
-    return srcNodeId;
-  }
-
-  public void setSrcNodeId(int srcNodeId) {
-    this.srcNodeId = srcNodeId;
-  }
-
-  public int getTargetNodeId() {
-    return targetNodeId;
-  }
-
-  public void setTargetNodeId(int targetNodeId) {
-    this.targetNodeId = targetNodeId;
-  }
-
-  public Partition getPartition() {
-    return partition;
-  }
-
-  public void setPartition(Partition partition) {
-    this.partition = partition;
-  }
-
-  public String getStream() {
-    return "stream:" + srcNodeId + "-" + targetNodeId;
-  }
-}
@@ -1,99 +0,0 @@
-package io.ray.streaming.runtime.core.graph;
-
-import io.ray.api.BaseActorHandle;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
-
-/**
- * Physical execution graph.
- *
- * <p>Notice: Temporary implementation for now to keep functional. This will be changed to
- * {@link ExecutionGraph} later when new stream task implementation is ready.
- */
-public class ExecutionGraph implements Serializable {
-  private long buildTime;
-  private List<ExecutionNode> executionNodeList;
-  private List<BaseActorHandle> sourceWorkers = new ArrayList<>();
-  private List<BaseActorHandle> sinkWorkers = new ArrayList<>();
-
-  public ExecutionGraph(List<ExecutionNode> executionNodes) {
-    this.executionNodeList = executionNodes;
-    for (ExecutionNode executionNode : executionNodeList) {
-      if (executionNode.getNodeType() == ExecutionNode.NodeType.SOURCE) {
-        List<BaseActorHandle> actors = executionNode.getExecutionTasks().stream()
-            .map(ExecutionTask::getWorker).collect(Collectors.toList());
-        sourceWorkers.addAll(actors);
-      }
-      if (executionNode.getNodeType() == ExecutionNode.NodeType.SINK) {
-        List<BaseActorHandle> actors = executionNode.getExecutionTasks().stream()
-            .map(ExecutionTask::getWorker).collect(Collectors.toList());
-        sinkWorkers.addAll(actors);
-      }
-    }
-    buildTime = System.currentTimeMillis();
-  }
-
-  public List<BaseActorHandle> getSourceWorkers() {
-    return sourceWorkers;
-  }
-
-  public List<BaseActorHandle> getSinkWorkers() {
-    return sinkWorkers;
-  }
-
-  public List<ExecutionNode> getExecutionNodeList() {
-    return executionNodeList;
-  }
-
-  public ExecutionTask getExecutionTaskByTaskId(int taskId) {
-    for (ExecutionNode executionNode : executionNodeList) {
-      for (ExecutionTask executionTask : executionNode.getExecutionTasks()) {
-        if (executionTask.getTaskId() == taskId) {
-          return executionTask;
-        }
-      }
-    }
-    throw new RuntimeException("Task " + taskId + " does not exist!");
-  }
-
-  public ExecutionNode getExecutionNodeByNodeId(int nodeId) {
-    for (ExecutionNode executionNode : executionNodeList) {
-      if (executionNode.getNodeId() == nodeId) {
-        return executionNode;
-      }
-    }
-    throw new RuntimeException("Node " + nodeId + " does not exist!");
-  }
-
-  public ExecutionNode getExecutionNodeByTaskId(int taskId) {
-    for (ExecutionNode executionNode : executionNodeList) {
-      for (ExecutionTask executionTask : executionNode.getExecutionTasks()) {
-        if (executionTask.getTaskId() == taskId) {
-          return executionNode;
-        }
-      }
-    }
-    throw new RuntimeException("Task " + taskId + " does not exist!");
-  }
-
-  public Map<Integer, BaseActorHandle> getTaskId2WorkerByNodeId(int nodeId) {
-    for (ExecutionNode executionNode : executionNodeList) {
-      if (executionNode.getNodeId() == nodeId) {
-        Map<Integer, BaseActorHandle> taskId2Worker = new HashMap<>();
-        for (ExecutionTask executionTask : executionNode.getExecutionTasks()) {
-          taskId2Worker.put(executionTask.getTaskId(), executionTask.getWorker());
-        }
-        return taskId2Worker;
-      }
-    }
-    throw new RuntimeException("Node " + nodeId + " does not exist!");
-  }
-
-  public long getBuildTime() {
-    return buildTime;
-  }
-}
@@ -1,126 +0,0 @@
-package io.ray.streaming.runtime.core.graph;
-
-import io.ray.streaming.api.Language;
-import io.ray.streaming.jobgraph.VertexType;
-import io.ray.streaming.operator.StreamOperator;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-/**
- * A node in the physical execution graph.
- */
-public class ExecutionNode implements Serializable {
-  private int nodeId;
-  private int parallelism;
-  private Map<String, String> config;
-  private NodeType nodeType;
-  private StreamOperator streamOperator;
-  private List<ExecutionTask> executionTasks;
-  private List<ExecutionEdge> inputsEdges;
-  private List<ExecutionEdge> outputEdges;
-
-  public ExecutionNode(int nodeId, int parallelism, Map<String, String> config) {
-    this.nodeId = nodeId;
-    this.parallelism = parallelism;
-    this.config = config;
-    this.executionTasks = new ArrayList<>();
-    this.inputsEdges = new ArrayList<>();
-    this.outputEdges = new ArrayList<>();
-  }
-
-  public int getNodeId() {
-    return nodeId;
-  }
-
-  public void setNodeId(int nodeId) {
-    this.nodeId = nodeId;
-  }
-
-  public int getParallelism() {
-    return parallelism;
-  }
-
-  public void setParallelism(int parallelism) {
-    this.parallelism = parallelism;
-  }
-
-  public Map<String, String> getConfig() {
-    return config;
-  }
-
-  public List<ExecutionTask> getExecutionTasks() {
-    return executionTasks;
-  }
-
-  public void setExecutionTasks(List<ExecutionTask> executionTasks) {
-    this.executionTasks = executionTasks;
-  }
-
-  public List<ExecutionEdge> getOutputEdges() {
-    return outputEdges;
-  }
-
-  public void setOutputEdges(List<ExecutionEdge> outputEdges) {
-    this.outputEdges = outputEdges;
-  }
-
-  public void addOutputEdge(ExecutionEdge executionEdge) {
-    this.outputEdges.add(executionEdge);
-  }
-
-  public void addInputEdge(ExecutionEdge executionEdge) {
-    this.inputsEdges.add(executionEdge);
-  }
-
-  public List<ExecutionEdge> getInputsEdges() {
-    return inputsEdges;
-  }
-
-  public StreamOperator getStreamOperator() {
-    return streamOperator;
-  }
-
-  public void setStreamOperator(StreamOperator streamOperator) {
-    this.streamOperator = streamOperator;
-  }
-
-  public Language getLanguage() {
-    return streamOperator.getLanguage();
-  }
-
-  public NodeType getNodeType() {
-    return nodeType;
-  }
-
-  public void setNodeType(VertexType vertexType) {
-    switch (vertexType) {
-      case SOURCE:
-        this.nodeType = NodeType.SOURCE;
-        break;
-      case SINK:
-        this.nodeType = NodeType.SINK;
-        break;
-      default:
-        this.nodeType = NodeType.TRANSFORM;
-    }
-  }
-
-  @Override
-  public String toString() {
-    final StringBuilder sb = new StringBuilder("ExecutionNode{");
-    sb.append("nodeId=").append(nodeId);
-    sb.append(", parallelism=").append(parallelism);
-    sb.append(", nodeType=").append(nodeType);
-    sb.append(", streamOperator=").append(streamOperator);
-    sb.append('}');
-    return sb.toString();
-  }
-
-  public enum NodeType {
-    SOURCE,
-    TRANSFORM,
-    SINK,
-  }
-}
@@ -1,45 +0,0 @@
-package io.ray.streaming.runtime.core.graph;
-
-import io.ray.api.BaseActorHandle;
-import java.io.Serializable;
-
-/**
- * ExecutionTask is minimal execution unit.
- * <p>
- * An ExecutionNode has n ExecutionTasks if parallelism is n.
- */
-public class ExecutionTask implements Serializable {
-  private int taskId;
-  private int taskIndex;
-  private BaseActorHandle worker;
-
-  public ExecutionTask(int taskId, int taskIndex, BaseActorHandle worker) {
-    this.taskId = taskId;
-    this.taskIndex = taskIndex;
-    this.worker = worker;
-  }
-
-  public int getTaskId() {
-    return taskId;
-  }
-
-  public void setTaskId(int taskId) {
-    this.taskId = taskId;
-  }
-
-  public int getTaskIndex() {
-    return taskIndex;
-  }
-
-  public void setTaskIndex(int taskIndex) {
-    this.taskIndex = taskIndex;
-  }
-
-  public BaseActorHandle getWorker() {
-    return worker;
-  }
-
-  public void setWorker(BaseActorHandle worker) {
-    this.worker = worker;
-  }
-}
@@ -12,12 +12,12 @@ public class ExecutionEdge implements Serializable {
  /**
   * The source(upstream) execution vertex.
   */
-  private final ExecutionVertex sourceVertex;
+  private final ExecutionVertex sourceExecutionVertex;

  /**
   * The target(downstream) execution vertex.
   */
-  private final ExecutionVertex targetVertex;
+  private final ExecutionVertex targetExecutionVertex;

  /**
   * The partition of current execution edge's execution job edge.
@@ -29,32 +29,35 @@ public class ExecutionEdge implements Serializable {
   */
  private final String executionEdgeIndex;

-  public ExecutionEdge(ExecutionVertex sourceVertex, ExecutionVertex targetVertex,
+  public ExecutionEdge(
+      ExecutionVertex sourceExecutionVertex,
+      ExecutionVertex targetExecutionVertex,
      ExecutionJobEdge executionJobEdge) {
-    this.sourceVertex = sourceVertex;
-    this.targetVertex = targetVertex;
+    this.sourceExecutionVertex = sourceExecutionVertex;
+    this.targetExecutionVertex = targetExecutionVertex;
    this.partition = executionJobEdge.getPartition();
    this.executionEdgeIndex = generateExecutionEdgeIndex();
  }

  private String generateExecutionEdgeIndex() {
-    return sourceVertex.getId() + "—" + targetVertex.getId();
+    return sourceExecutionVertex.getExecutionVertexId() + "—"
+        + targetExecutionVertex.getExecutionVertexId();
  }

-  public ExecutionVertex getSourceVertex() {
-    return sourceVertex;
+  public ExecutionVertex getSourceExecutionVertex() {
+    return sourceExecutionVertex;
  }

-  public ExecutionVertex getTargetVertex() {
-    return targetVertex;
+  public ExecutionVertex getTargetExecutionVertex() {
+    return targetExecutionVertex;
  }

  public int getSourceVertexId() {
-    return sourceVertex.getId();
+    return sourceExecutionVertex.getExecutionVertexId();
  }

  public int getTargetVertexId() {
-    return targetVertex.getId();
+    return targetExecutionVertex.getExecutionVertexId();
  }

  public Partition getPartition() {
@@ -68,10 +71,10 @@ public class ExecutionEdge implements Serializable {
  @Override
  public String toString() {
    return MoreObjects.toStringHelper(this)
-      .add("sourceVertex", sourceVertex)
-      .add("targetVertex", targetVertex)
+      .add("source", sourceExecutionVertex)
+      .add("target", targetExecutionVertex)
      .add("partition", partition)
-      .add("executionEdgeIndex", executionEdgeIndex)
+      .add("index", executionEdgeIndex)
      .toString();
  }
 }
@@ -1,7 +1,6 @@
 package io.ray.streaming.runtime.core.graph.executiongraph;

-import io.ray.api.ActorHandle;
-import io.ray.streaming.runtime.worker.JobWorker;
+import io.ray.api.BaseActorHandle;
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -130,7 +129,7 @@ public class ExecutionGraph implements Serializable {
  public ExecutionVertex getExecutionJobVertexByJobVertexId(int vertexId) {
    for (ExecutionJobVertex executionJobVertex : executionJobVertexMap.values()) {
      for (ExecutionVertex executionVertex : executionJobVertex.getExecutionVertices()) {
-        if (executionVertex.getId() == vertexId) {
+        if (executionVertex.getExecutionVertexId() == vertexId) {
          return executionVertex;
        }
      }
@@ -143,7 +142,7 @@ public class ExecutionGraph implements Serializable {
   *
   * @return actor list
   */
-  public List<ActorHandle<JobWorker>> getAllActors() {
+  public List<BaseActorHandle> getAllActors() {
    return getActorsFromJobVertices(getExecutionJobVertexList());
  }

@@ -152,7 +151,7 @@ public class ExecutionGraph implements Serializable {
   *
   * @return actor list
   */
-  public List<ActorHandle<JobWorker>> getSourceActors() {
+  public List<BaseActorHandle> getSourceActors() {
    List<ExecutionJobVertex> executionJobVertices = getExecutionJobVertexList().stream()
        .filter(ExecutionJobVertex::isSourceVertex)
        .collect(Collectors.toList());
@@ -165,7 +164,7 @@ public class ExecutionGraph implements Serializable {
   *
   * @return actor list
   */
-  public List<ActorHandle<JobWorker>> getNonSourceActors() {
+  public List<BaseActorHandle> getNonSourceActors() {
    List<ExecutionJobVertex> executionJobVertices = getExecutionJobVertexList().stream()
        .filter(executionJobVertex -> executionJobVertex.isTransformationVertex()
            || executionJobVertex.isSinkVertex())
@@ -179,7 +178,7 @@ public class ExecutionGraph implements Serializable {
   *
   * @return actor list
   */
-  public List<ActorHandle<JobWorker>> getSinkActors() {
+  public List<BaseActorHandle> getSinkActors() {
    List<ExecutionJobVertex> executionJobVertices = getExecutionJobVertexList().stream()
        .filter(ExecutionJobVertex::isSinkVertex)
        .collect(Collectors.toList());
@@ -193,7 +192,7 @@ public class ExecutionGraph implements Serializable {
   * @param executionJobVertices specified job vertices
   * @return actor list
   */
-  public List<ActorHandle<JobWorker>> getActorsFromJobVertices(
+  public List<BaseActorHandle> getActorsFromJobVertices(
      List<ExecutionJobVertex> executionJobVertices) {
    return executionJobVertices.stream()
        .map(ExecutionJobVertex::getExecutionVertices)
@@ -12,12 +12,12 @@ public class ExecutionJobEdge {
  /**
   * The source(upstream) execution job vertex.
   */
-  private final ExecutionJobVertex sourceVertex;
+  private final ExecutionJobVertex sourceExecutionJobVertex;

  /**
   * The target(downstream) execution job vertex.
   */
-  private final ExecutionJobVertex targetVertex;
+  private final ExecutionJobVertex targetExecutionJobVertex;

  /**
   * The partition of the execution job edge.
@@ -29,24 +29,27 @@ public class ExecutionJobEdge {
   */
  private final String executionJobEdgeIndex;

-  public ExecutionJobEdge(ExecutionJobVertex sourceVertex, ExecutionJobVertex targetVertex,
+  public ExecutionJobEdge(
+      ExecutionJobVertex sourceExecutionJobVertex,
+      ExecutionJobVertex targetExecutionJobVertex,
      JobEdge jobEdge) {
-    this.sourceVertex = sourceVertex;
-    this.targetVertex = targetVertex;
+    this.sourceExecutionJobVertex = sourceExecutionJobVertex;
+    this.targetExecutionJobVertex = targetExecutionJobVertex;
    this.partition = jobEdge.getPartition();
    this.executionJobEdgeIndex = generateExecutionJobEdgeIndex();
  }

  private String generateExecutionJobEdgeIndex() {
-    return sourceVertex.getJobVertexId() + "—" + targetVertex.getJobVertexId();
+    return sourceExecutionJobVertex.getExecutionJobVertexId() + "—"
+        + targetExecutionJobVertex.getExecutionJobVertexId();
  }

-  public ExecutionJobVertex getSourceVertex() {
-    return sourceVertex;
+  public ExecutionJobVertex getSourceExecutionJobVertex() {
+    return sourceExecutionJobVertex;
  }

-  public ExecutionJobVertex getTargetVertex() {
-    return targetVertex;
+  public ExecutionJobVertex getTargetExecutionJobVertex() {
+    return targetExecutionJobVertex;
  }

  public Partition getPartition() {
@@ -56,10 +59,10 @@ public class ExecutionJobEdge {
  @Override
  public String toString() {
    return MoreObjects.toStringHelper(this)
-        .add("srcVertex", sourceVertex)
-        .add("targetVertex", targetVertex)
+        .add("source", sourceExecutionJobVertex)
+        .add("target", targetExecutionJobVertex)
        .add("partition", partition)
-        .add("executionJobEdgeIndex", executionJobEdgeIndex)
+        .add("index", executionJobEdgeIndex)
        .toString();
  }
 }
@@ -2,13 +2,12 @@ package io.ray.streaming.runtime.core.graph.executiongraph;

 import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
-import io.ray.api.ActorHandle;
+import io.ray.api.BaseActorHandle;
 import io.ray.streaming.api.Language;
 import io.ray.streaming.jobgraph.JobVertex;
 import io.ray.streaming.jobgraph.VertexType;
 import io.ray.streaming.operator.StreamOperator;
 import io.ray.streaming.runtime.config.master.ResourceConfig;
-import io.ray.streaming.runtime.worker.JobWorker;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -25,18 +24,20 @@ import org.aeonbits.owner.ConfigFactory;
 public class ExecutionJobVertex {

  /**
-   * Unique id of operator(use {@link JobVertex}'s id). Used as jobVertex's id.
+   * Unique id. Use {@link JobVertex}'s id directly.
   */
-  private final int jobVertexId;
+  private final int executionJobVertexId;

  /**
-   * Unique name of operator(use {@link StreamOperator}'s name). Used as jobVertex's name.
+   * Use jobVertex id and operator(use {@link StreamOperator}'s name) as name.
+   * e.g. 1-SourceOperator
   */
-  private final String jobVertexName;
+  private final String executionJobVertexName;
  private final StreamOperator streamOperator;
  private final VertexType vertexType;
  private final Language language;
  private final Map<String, String> jobConfig;
+  private final long buildTime;

  /**
   * Parallelism of current execution job vertex(operator).
@@ -55,18 +56,23 @@ public class ExecutionJobVertex {
  private List<ExecutionJobEdge> outputEdges = new ArrayList<>();

  public ExecutionJobVertex(
-      JobVertex jobVertex, Map<String, String> jobConfig, AtomicInteger idGenerator) {
-    this.jobVertexId = jobVertex.getVertexId();
-    this.jobVertexName = jobVertex.getStreamOperator().getName();
+      JobVertex jobVertex,
+      Map<String, String> jobConfig,
+      AtomicInteger idGenerator,
+      long buildTime) {
+    this.executionJobVertexId = jobVertex.getVertexId();
+    this.executionJobVertexName = generateExecutionJobVertexName(
+        executionJobVertexId, jobVertex.getStreamOperator().getName());
    this.streamOperator = jobVertex.getStreamOperator();
    this.vertexType = jobVertex.getVertexType();
    this.language = jobVertex.getLanguage();
    this.jobConfig = jobConfig;
+    this.buildTime = buildTime;
    this.parallelism = jobVertex.getParallelism();
-    this.executionVertices = createExecutionVertics(idGenerator);
+    this.executionVertices = createExecutionVertices(idGenerator);
  }

-  private List<ExecutionVertex> createExecutionVertics(AtomicInteger idGenerator) {
+  private List<ExecutionVertex> createExecutionVertices(AtomicInteger idGenerator) {
    List<ExecutionVertex> executionVertices = new ArrayList<>();
    ResourceConfig resourceConfig = ConfigFactory.create(ResourceConfig.class, jobConfig);

@@ -77,8 +83,12 @@ public class ExecutionJobVertex {
    return executionVertices;
  }

-  public Map<Integer, ActorHandle<JobWorker>> getExecutionVertexWorkers() {
-    Map<Integer, ActorHandle<JobWorker>> executionVertexWorkersMap = new HashMap<>();
+  private String generateExecutionJobVertexName(int jobVertexId, String streamOperatorName) {
+    return jobVertexId + "-" + streamOperatorName;
+  }
+
+  public Map<Integer, BaseActorHandle> getExecutionVertexWorkers() {
+    Map<Integer, BaseActorHandle> executionVertexWorkersMap = new HashMap<>();

    Preconditions.checkArgument(
        executionVertices != null && !executionVertices.isEmpty(),
@@ -87,18 +97,18 @@ public class ExecutionJobVertex {
      Preconditions.checkArgument(
          vertex.getWorkerActor() != null,
          "Empty execution vertex worker actor.");
-      executionVertexWorkersMap.put(vertex.getId(), vertex.getWorkerActor());
+      executionVertexWorkersMap.put(vertex.getExecutionVertexId(), vertex.getWorkerActor());
    });

    return executionVertexWorkersMap;
  }

-  public int getJobVertexId() {
-    return jobVertexId;
+  public int getExecutionJobVertexId() {
+    return executionJobVertexId;
  }

-  public String getJobVertexName() {
-    return jobVertexName;
+  public String getExecutionJobVertexName() {
+    return executionJobVertexName;
  }

  /**
@@ -106,8 +116,8 @@ public class ExecutionJobVertex {
   *
   * @return operator name with index
   */
-  public String getVertexNameWithIndex() {
-    return jobVertexId + "-" + jobVertexName;
+  public String getExecutionJobVertexNameWithIndex() {
+    return executionJobVertexId + "-" + executionJobVertexName;
  }

  public int getParallelism() {
@@ -153,6 +163,14 @@ public class ExecutionJobVertex {
    return language;
  }

+  public Map<String, String> getJobConfig() {
+    return jobConfig;
+  }
+
+  public long getBuildTime() {
+    return buildTime;
+  }
+
  public boolean isSourceVertex() {
    return getVertexType() == VertexType.SOURCE;
  }
@@ -168,8 +186,8 @@ public class ExecutionJobVertex {
  @Override
  public String toString() {
    return MoreObjects.toStringHelper(this)
-        .add("jobVertexId", jobVertexId)
-        .add("jobVertexName", jobVertexName)
+        .add("executionJobVertexId", executionJobVertexId)
+        .add("executionJobVertexName", executionJobVertexName)
        .add("vertexType", vertexType)
        .add("parallelism", parallelism)
        .toString();
@@ -1,7 +1,7 @@
 package io.ray.streaming.runtime.core.graph.executiongraph;

 import com.google.common.base.MoreObjects;
-import io.ray.api.ActorHandle;
+import io.ray.api.BaseActorHandle;
 import io.ray.api.id.ActorId;
 import io.ray.streaming.api.Language;
 import io.ray.streaming.jobgraph.VertexType;
@@ -9,13 +9,13 @@ import io.ray.streaming.operator.StreamOperator;
 import io.ray.streaming.runtime.config.master.ResourceConfig;
 import io.ray.streaming.runtime.core.resource.ContainerID;
 import io.ray.streaming.runtime.core.resource.ResourceType;
-import io.ray.streaming.runtime.worker.JobWorker;
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.stream.Collectors;

 /**
 * Physical vertex, correspond to {@link ExecutionJobVertex}.
@@ -25,31 +25,51 @@ public class ExecutionVertex implements Serializable {
  /**
   * Unique id for execution vertex.
   */
-  private final int id;
+  private final int executionVertexId;

  /**
   * Immutable field inherited from {@link ExecutionJobVertex}.
   */
-  private final int jobVertexId;
-  private final String jobVertexName;
+  private final int executionJobVertexId;
+  private final String executionJobVertexName;
  private final StreamOperator streamOperator;
  private final VertexType vertexType;
  private final Language language;
+  private final long buildTime;

  /**
-   * Resources used by ExecutionVertex.
+   * Resource used by ExecutionVertex.
   */
-  private final Map<String, Double> resources;
+  private final Map<String, Double> resource;
+
+  /**
+   * Parallelism of current vertex's operator.
+   */
+  private int parallelism;

  /**
   * Ordered sub index for execution vertex in a execution job vertex.
   * Might be changed in dynamic scheduling.
   */
-  private int vertexIndex;
+  private int executionVertexIndex;

  private ExecutionVertexState state = ExecutionVertexState.TO_ADD;
+
+  /**
+   * The id of the container which this vertex's worker actor belongs to.
+   */
  private ContainerID containerId;
-  private ActorHandle<JobWorker> workerActor;
+
+  /**
+   * Worker actor handle.
+   */
+  private BaseActorHandle workerActor;
+
+  /**
+   * Op config + job config.
+   */
+  private Map<String, String> workerConfig;
+
  private List<ExecutionEdge> inputEdges = new ArrayList<>();
  private List<ExecutionEdge> outputEdges = new ArrayList<>();

@@ -58,26 +78,43 @@ public class ExecutionVertex implements Serializable {
      int index,
      ExecutionJobVertex executionJobVertex,
      ResourceConfig resourceConfig) {
-    this.id = globalIndex;
-    this.jobVertexId = executionJobVertex.getJobVertexId();
-    this.jobVertexName = executionJobVertex.getJobVertexName();
+    this.executionVertexId = globalIndex;
+    this.executionJobVertexId = executionJobVertex.getExecutionJobVertexId();
+    this.executionJobVertexName = executionJobVertex.getExecutionJobVertexName();
    this.streamOperator = executionJobVertex.getStreamOperator();
    this.vertexType = executionJobVertex.getVertexType();
    this.language = executionJobVertex.getLanguage();
-    this.vertexIndex = index;
-    this.resources = generateResources(resourceConfig);
+    this.buildTime = executionJobVertex.getBuildTime();
+    this.parallelism = executionJobVertex.getParallelism();
+    this.executionVertexIndex = index;
+    this.resource = generateResources(resourceConfig);
+    this.workerConfig = genWorkerConfig(executionJobVertex.getJobConfig());
  }

-  public int getId() {
-    return id;
+  private Map<String, String> genWorkerConfig(Map<String, String> jobConfig) {
+    Map<String, String> workerConfig = new HashMap<>();
+    workerConfig.putAll(jobConfig);
+    return workerConfig;
  }

-  public int getJobVertexId() {
-    return jobVertexId;
+  public int getExecutionVertexId() {
+    return executionVertexId;
  }

-  public String getJobVertexName() {
-    return jobVertexName;
+  /**
+   * Unique name generated by execution job vertex name and index of current execution vertex.
+   * e.g. 1-SourceOperator-3 (vertex index is 3)
+   */
+  public String getExecutionVertexName() {
+    return executionJobVertexName + "-" + executionVertexIndex;
+  }
+
+  public int getExecutionJobVertexId() {
+    return executionJobVertexId;
+  }
+
+  public String getExecutionJobVertexName() {
+    return executionJobVertexName;
  }

  public StreamOperator getStreamOperator() {
@@ -92,16 +129,12 @@ public class ExecutionVertex implements Serializable {
    return language;
  }

-  public int getVertexIndex() {
-    return vertexIndex;
+  public int getParallelism() {
+    return parallelism;
  }

-  /**
-   * Unique name generated by vertex name and index for execution vertex.
-   * e.g. 1-SourceOperator-3 (vertex index is 3)
-   */
-  public String getVertexName() {
-    return jobVertexId + "-" + jobVertexName + "-" + vertexIndex;
+  public int getExecutionVertexIndex() {
+    return executionVertexIndex;
  }

  public ExecutionVertexState getState() {
@@ -124,7 +157,7 @@ public class ExecutionVertex implements Serializable {
    return state == ExecutionVertexState.TO_DEL;
  }

-  public ActorHandle<JobWorker> getWorkerActor() {
+  public BaseActorHandle getWorkerActor() {
    return workerActor;
  }

@@ -132,7 +165,7 @@ public class ExecutionVertex implements Serializable {
    return workerActor.getId();
  }

-  public void setWorkerActor(ActorHandle<JobWorker> workerActor) {
+  public void setWorkerActor(BaseActorHandle workerActor) {
    this.workerActor = workerActor;
  }

@@ -154,8 +187,28 @@ public class ExecutionVertex implements Serializable {
    this.outputEdges = outputEdges;
  }

-  public Map<String, Double> getResources() {
-    return resources;
+  public List<ExecutionVertex> getInputVertices() {
+    return inputEdges.stream()
+        .map(ExecutionEdge::getSourceExecutionVertex)
+        .collect(Collectors.toList());
+  }
+
+  public List<ExecutionVertex> getOutputVertices() {
+    return outputEdges.stream()
+        .map(ExecutionEdge::getTargetExecutionVertex)
+        .collect(Collectors.toList());
+  }
+
+  public Map<String, Double> getResource() {
+    return resource;
+  }
+
+  public Map<String, String> getWorkerConfig() {
+    return workerConfig;
+  }
+
+  public long getBuildTime() {
+    return buildTime;
  }

  public ContainerID getContainerId() {
@@ -186,22 +239,22 @@ public class ExecutionVertex implements Serializable {
  @Override
  public boolean equals(Object obj) {
    if (obj instanceof ExecutionVertex) {
-      return this.id == ((ExecutionVertex)obj).getId();
+      return this.executionVertexId == ((ExecutionVertex)obj).getExecutionVertexId();
    }
    return false;
  }

  @Override
  public int hashCode() {
-    return Objects.hash(id, outputEdges);
+    return Objects.hash(executionVertexId, outputEdges);
  }

  @Override
  public String toString() {
    return MoreObjects.toStringHelper(this)
-        .add("id", id)
-        .add("name", getVertexName())
-        .add("resources", resources)
+        .add("id", executionVertexId)
+        .add("name", getExecutionVertexName())
+        .add("resources", resource)
        .add("state", state)
        .add("containerId", containerId)
        .add("workerActor", workerActor)
@@ -138,21 +138,21 @@ public class Container implements Serializable {
  public void allocateActor(ExecutionVertex vertex) {
    LOG.info("Allocating vertex [{}] in container [{}].", vertex, this);

-    executionVertexIds.add(vertex.getId());
+    executionVertexIds.add(vertex.getExecutionVertexId());
    vertex.setContainerIfNotExist(this.getId());
    // Binding dynamic resource
-    vertex.getResources().put(getName(), 1.0);
-    decreaseResource(vertex.getResources());
+    vertex.getResource().put(getName(), 1.0);
+    decreaseResource(vertex.getResource());
  }

  public void releaseActor(ExecutionVertex vertex) {
    LOG.info("Release actor, vertex: {}, container: {}.", vertex, vertex.getContainerId());
-    if (executionVertexIds.contains(vertex.getId())) {
-      executionVertexIds.removeIf(id -> id == vertex.getId());
-      reclaimResource(vertex.getResources());
+    if (executionVertexIds.contains(vertex.getExecutionVertexId())) {
+      executionVertexIds.removeIf(id -> id == vertex.getExecutionVertexId());
+      reclaimResource(vertex.getResource());
    } else {
      throw new RuntimeException(String.format("Current container [%s] not found vertex [%s].",
-          this, vertex.getJobVertexName()));
+          this, vertex.getExecutionJobVertexName()));
    }
  }

@@ -30,7 +30,7 @@ public class JobMaster {
  private GraphManager graphManager;
  private StreamingMasterConfig conf;

-  private ActorHandle jobMasterActor;
+  private ActorHandle<JobMaster> jobMasterActor;

  public JobMaster(Map<String, String> confMap) {
    LOG.info("Creating job master with conf: {}.", confMap);
@@ -101,7 +101,7 @@ public class JobMaster {
    return true;
  }

-  public ActorHandle getJobMasterActor() {
+  public ActorHandle<JobMaster> getJobMasterActor() {
    return jobMasterActor;
  }

@@ -53,7 +53,10 @@ public class GraphManagerImpl implements GraphManager {
      int jobVertexId = jobVertex.getVertexId();
      exeJobVertexMap.put(jobVertexId,
          new ExecutionJobVertex(
-              jobVertex, jobConfig, executionGraph.getExecutionVertexIdGenerator()));
+              jobVertex,
+              jobConfig,
+              executionGraph.getExecutionVertexIdGenerator(),
+              buildTime));
    }

    // connect vertex
@@ -84,7 +84,7 @@ public class PipelineFirstStrategy implements ResourceAssignStrategy {
          continue;
        }
        ExecutionVertex executionVertex = exeVertices.get(i);
-        Map<String, Double> requiredResource = executionVertex.getResources();
+        Map<String, Double> requiredResource = executionVertex.getResource();
        if (requiredResource.containsKey(ResourceType.CPU.getValue())) {
          LOG.info("Required resource contain {} value : {}, no limitation by default.",
              ResourceType.CPU, requiredResource.get(ResourceType.CPU.getValue()));
@@ -153,29 +153,28 @@ public class JobSchedulerImpl implements JobScheduler {
   */
  protected Map<ExecutionVertex, JobWorkerContext> buildWorkersContext(
      ExecutionGraph executionGraph) {
-    ActorHandle masterActor = jobMaster.getJobMasterActor();
+    ActorHandle<JobMaster> masterActor = jobMaster.getJobMasterActor();

    // build workers' context
-    Map<ExecutionVertex, JobWorkerContext> needRegistryVertexToContextMap = new HashMap<>();
+    Map<ExecutionVertex, JobWorkerContext> vertexToContextMap = new HashMap<>();
    executionGraph.getAllExecutionVertices().forEach(vertex -> {
-      JobWorkerContext ctx = buildJobWorkerContext(vertex, masterActor);
-      needRegistryVertexToContextMap.put(vertex, ctx);
+      JobWorkerContext context = buildJobWorkerContext(vertex, masterActor);
+      vertexToContextMap.put(vertex, context);
    });
-    return needRegistryVertexToContextMap;
+    return vertexToContextMap;
  }

  private JobWorkerContext buildJobWorkerContext(
      ExecutionVertex executionVertex,
      ActorHandle<JobMaster> masterActor) {

-    // create worker context
-    JobWorkerContext ctx = new JobWorkerContext(
-        executionVertex.getWorkerActorId(),
+    // create java worker context
+    JobWorkerContext context = new JobWorkerContext(
        masterActor,
        executionVertex
    );

-    return ctx;
+    return context;
  }

  /**
@@ -1,9 +1,10 @@
 package io.ray.streaming.runtime.master.scheduler.controller;

-import io.ray.api.ActorHandle;
+import io.ray.api.BaseActorHandle;
 import io.ray.api.ObjectRef;
 import io.ray.api.Ray;
 import io.ray.api.WaitResult;
+import io.ray.api.function.PyActorClass;
 import io.ray.api.id.ActorId;
 import io.ray.api.options.ActorCreationOptions;
 import io.ray.streaming.api.Language;
@@ -41,17 +42,22 @@ public class WorkerLifecycleController {
   */
  private boolean createWorker(ExecutionVertex executionVertex) {
    LOG.info("Start to create worker actor for vertex: {} with resource: {}.",
-        executionVertex.getVertexName(), executionVertex.getResources());
+        executionVertex.getExecutionVertexName(), executionVertex.getResource());

    Language language = executionVertex.getLanguage();

    ActorCreationOptions options = new ActorCreationOptions.Builder()
-        .setResources(executionVertex.getResources())
+        .setResources(executionVertex.getResource())
        .setMaxRestarts(-1)
        .createActorCreationOptions();

-    ActorHandle<JobWorker> actor = null;
-    // TODO (datayjz): ray create actor
+    BaseActorHandle actor;
+    if (Language.JAVA == language) {
+      actor = Ray.createActor(JobWorker::new, options);
+    } else {
+      actor = Ray.createActor(
+          new PyActorClass("ray.streaming.runtime.worker", "JobWorker"));
+    }

    if (null == actor) {
      LOG.error("Create worker actor failed.");
@@ -61,7 +67,7 @@ public class WorkerLifecycleController {
    executionVertex.setWorkerActor(actor);

    LOG.info("Worker actor created, actor: {}, vertex: {}.",
-        executionVertex.getWorkerActorId(), executionVertex.getVertexName());
+        executionVertex.getWorkerActorId(), executionVertex.getExecutionVertexName());
    return true;
  }

@@ -139,15 +145,15 @@ public class WorkerLifecycleController {
  }

  private boolean destroyWorker(ExecutionVertex executionVertex) {
-    ActorHandle rayActor = executionVertex.getWorkerActor();
+    BaseActorHandle rayActor = executionVertex.getWorkerActor();
    LOG.info("Begin destroying worker[vertex={}, actor={}].",
-        executionVertex.getVertexName(), rayActor.getId());
+        executionVertex.getExecutionVertexName(), rayActor.getId());

    boolean destroyResult = RemoteCallWorker.shutdownWithoutReconstruction(rayActor);

    if (!destroyResult) {
      LOG.error("Failed to destroy JobWorker[{}]'s actor: {}.",
-          executionVertex.getVertexName(), rayActor);
+          executionVertex.getExecutionVertexName(), rayActor);
      return false;
    }

@@ -8,70 +8,98 @@ import io.ray.streaming.operator.Operator;
 import io.ray.streaming.python.PythonFunction;
 import io.ray.streaming.python.PythonOperator;
 import io.ray.streaming.python.PythonPartition;
-import io.ray.streaming.runtime.core.graph.ExecutionEdge;
-import io.ray.streaming.runtime.core.graph.ExecutionGraph;
-import io.ray.streaming.runtime.core.graph.ExecutionNode;
-import io.ray.streaming.runtime.core.graph.ExecutionTask;
+import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionEdge;
+import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionVertex;
 import io.ray.streaming.runtime.generated.RemoteCall;
 import io.ray.streaming.runtime.generated.Streaming;
 import io.ray.streaming.runtime.serialization.MsgPackSerializer;
 import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;

 public class GraphPbBuilder {

  private MsgPackSerializer serializer = new MsgPackSerializer();

-  /**
-   * For simple scenario, a single ExecutionNode is enough. But some cases may need
-   * sub-graph information, so we serialize entire graph.
-   */
-  public RemoteCall.ExecutionGraph buildExecutionGraphPb(ExecutionGraph graph) {
-    RemoteCall.ExecutionGraph.Builder builder = RemoteCall.ExecutionGraph.newBuilder();
-    builder.setBuildTime(graph.getBuildTime());
-    for (ExecutionNode node : graph.getExecutionNodeList()) {
-      RemoteCall.ExecutionGraph.ExecutionNode.Builder nodeBuilder =
-          RemoteCall.ExecutionGraph.ExecutionNode.newBuilder();
-      nodeBuilder.setNodeId(node.getNodeId());
-      nodeBuilder.setParallelism(node.getParallelism());
-      nodeBuilder.setNodeType(
-          Streaming.NodeType.valueOf(node.getNodeType().name()));
-      nodeBuilder.setLanguage(Streaming.Language.valueOf(node.getLanguage().name()));
-      byte[] operatorBytes = serializeOperator(node.getStreamOperator());
-      nodeBuilder.setOperator(ByteString.copyFrom(operatorBytes));
+  public RemoteCall.ExecutionVertexContext buildExecutionVertexContext(
+      ExecutionVertex executionVertex) {
+    RemoteCall.ExecutionVertexContext.Builder builder =
+        RemoteCall.ExecutionVertexContext.newBuilder();

-      // build tasks
-      for (ExecutionTask task : node.getExecutionTasks()) {
-        RemoteCall.ExecutionGraph.ExecutionTask.Builder taskBuilder =
-            RemoteCall.ExecutionGraph.ExecutionTask.newBuilder();
-        byte[] serializedActorHandle = ((NativeActorHandle) task.getWorker()).toBytes();
-        taskBuilder
-            .setTaskId(task.getTaskId())
-            .setTaskIndex(task.getTaskIndex())
-            .setWorkerActor(ByteString.copyFrom(serializedActorHandle));
-        nodeBuilder.addExecutionTasks(taskBuilder.build());
-      }
+    // build vertex
+    builder.setCurrentExecutionVertex(buildVertex(executionVertex));

-      // build edges
-      for (ExecutionEdge edge : node.getInputsEdges()) {
-        nodeBuilder.addInputEdges(buildEdgePb(edge));
-      }
-      for (ExecutionEdge edge : node.getOutputEdges()) {
-        nodeBuilder.addOutputEdges(buildEdgePb(edge));
-      }
+    // build upstream vertices
+    List<ExecutionVertex> upstreamVertices = executionVertex.getInputVertices();
+    List<RemoteCall.ExecutionVertexContext.ExecutionVertex> upstreamVertexPbs =
+        upstreamVertices.stream()
+        .map(this::buildVertex)
+        .collect(Collectors.toList());
+    builder.addAllUpstreamExecutionVertices(upstreamVertexPbs);

-      builder.addExecutionNodes(nodeBuilder.build());
-    }
+    // build downstream vertices
+    List<ExecutionVertex> downstreamVertices = executionVertex.getOutputVertices();
+    List<RemoteCall.ExecutionVertexContext.ExecutionVertex> downstreamVertexPbs =
+        downstreamVertices.stream()
+            .map(this::buildVertex)
+            .collect(Collectors.toList());
+    builder.addAllDownstreamExecutionVertices(downstreamVertexPbs);
+
+    // build input edges
+    List<ExecutionEdge> inputEdges = executionVertex.getInputEdges();
+    List<RemoteCall.ExecutionVertexContext.ExecutionEdge> inputEdgesPbs =
+        inputEdges.stream()
+            .map(this::buildEdge)
+            .collect(Collectors.toList());
+    builder.addAllInputExecutionEdges(inputEdgesPbs);
+
+    // build output edges
+    List<ExecutionEdge> outputEdges = executionVertex.getOutputEdges();
+    List<RemoteCall.ExecutionVertexContext.ExecutionEdge> outputEdgesPbs =
+        outputEdges.stream()
+            .map(this::buildEdge)
+            .collect(Collectors.toList());
+    builder.addAllOutputExecutionEdges(outputEdgesPbs);

    return builder.build();
  }

-  private RemoteCall.ExecutionGraph.ExecutionEdge buildEdgePb(ExecutionEdge edge) {
-    RemoteCall.ExecutionGraph.ExecutionEdge.Builder edgeBuilder =
-        RemoteCall.ExecutionGraph.ExecutionEdge.newBuilder();
-    edgeBuilder.setSrcNodeId(edge.getSrcNodeId());
-    edgeBuilder.setTargetNodeId(edge.getTargetNodeId());
-    edgeBuilder.setPartition(ByteString.copyFrom(serializePartition(edge.getPartition())));
-    return edgeBuilder.build();
+  private RemoteCall.ExecutionVertexContext.ExecutionVertex buildVertex(
+      ExecutionVertex executionVertex) {
+    // build vertex infos
+    RemoteCall.ExecutionVertexContext.ExecutionVertex.Builder executionVertexBuilder =
+        RemoteCall.ExecutionVertexContext.ExecutionVertex.newBuilder();
+    executionVertexBuilder.setExecutionVertexId(executionVertex.getExecutionVertexId());
+    executionVertexBuilder.setExecutionJobVertexId(executionVertex.getExecutionJobVertexId());
+    executionVertexBuilder.setExecutionJobVertexName(executionVertex.getExecutionJobVertexName());
+    executionVertexBuilder.setExecutionVertexIndex(executionVertex.getExecutionVertexIndex());
+    executionVertexBuilder.setParallelism(executionVertex.getParallelism());
+    executionVertexBuilder.setOperator(
+        ByteString.copyFrom(
+            serializeOperator(executionVertex.getStreamOperator())));
+    executionVertexBuilder.setWorkerActor(
+        ByteString.copyFrom(
+            ((NativeActorHandle) (executionVertex.getWorkerActor())).toBytes()));
+    executionVertexBuilder.setContainerId(executionVertex.getContainerId().toString());
+    executionVertexBuilder.setBuildTime(executionVertex.getBuildTime());
+    executionVertexBuilder.setLanguage(
+        Streaming.Language.valueOf(executionVertex.getLanguage().name()));
+    executionVertexBuilder.putAllConfig(executionVertex.getWorkerConfig());
+    executionVertexBuilder.putAllResource(executionVertex.getResource());
+
+    return executionVertexBuilder.build();
+  }
+
+  private RemoteCall.ExecutionVertexContext.ExecutionEdge buildEdge(ExecutionEdge executionEdge) {
+    // build edge infos
+    RemoteCall.ExecutionVertexContext.ExecutionEdge.Builder executionEdgeBuilder =
+        RemoteCall.ExecutionVertexContext.ExecutionEdge.newBuilder();
+    executionEdgeBuilder.setSourceExecutionVertexId(executionEdge.getSourceVertexId());
+    executionEdgeBuilder.setTargetExecutionVertexId(executionEdge.getTargetVertexId());
+    executionEdgeBuilder.setPartition(
+        ByteString.copyFrom(serializePartition(executionEdge.getPartition())));
+
+    return executionEdgeBuilder.build();
  }

  private byte[] serializeOperator(Operator operator) {
@@ -1,7 +1,10 @@
 package io.ray.streaming.runtime.rpc;

 import io.ray.api.ActorHandle;
+import io.ray.api.BaseActorHandle;
 import io.ray.api.ObjectRef;
+import io.ray.api.PyActorHandle;
+import io.ray.api.function.PyActorMethod;
 import io.ray.streaming.runtime.master.JobMaster;
 import io.ray.streaming.runtime.worker.JobWorker;
 import io.ray.streaming.runtime.worker.context.JobWorkerContext;
@@ -20,16 +23,23 @@ public class RemoteCallWorker {
   * Call JobWorker actor to init.
   *
   * @param actor target JobWorker actor
-   * @param ctx JobWorker's context
+   * @param context JobWorker's context
   * @return init result
   */
-  public static ObjectRef<Boolean> initWorker(ActorHandle actor, JobWorkerContext ctx) {
-    LOG.info("Call worker to init, actor: {}, context: {}.", actor.getId(), ctx);
-    ObjectRef<Boolean> result = null;
+  public static ObjectRef<Boolean> initWorker(BaseActorHandle actor, JobWorkerContext context) {
+    LOG.info("Call worker to initiate, actor: {}, context: {}.", actor.getId(), context);
+    ObjectRef<Boolean> result;

-    // TODO (datayjz): ray call worker to initiate
+    // python
+    if (actor instanceof PyActorHandle) {
+      result = ((PyActorHandle) actor).call(
+          new PyActorMethod("init", Object.class), context.getPythonWorkerContextBytes());
+    } else {
+      // java
+      result = ((ActorHandle<JobWorker>) actor).call(JobWorker::init, context);
+    }

-    LOG.info("Finished calling worker to init.");
+    LOG.info("Finished calling worker to initiate.");
    return result;
  }

@@ -39,11 +49,17 @@ public class RemoteCallWorker {
   * @param actor target JobWorker actor
   * @return start result
   */
-  public static ObjectRef<Boolean> startWorker(ActorHandle actor) {
+  public static ObjectRef<Boolean> startWorker(BaseActorHandle actor) {
    LOG.info("Call worker to start, actor: {}.", actor.getId());
    ObjectRef<Boolean> result = null;

-    // TODO (datayjz): ray call worker to start
+    // python
+    if (actor instanceof PyActorHandle) {
+      result = ((PyActorHandle) actor).call(new PyActorMethod("start", Object.class));
+    } else {
+      // java
+      result = ((ActorHandle<JobWorker>) actor).call(JobWorker::start);
+    }

    LOG.info("Finished calling worker to start.");
    return result;
@@ -55,7 +71,7 @@ public class RemoteCallWorker {
   * @param actor target JobWorker actor
   * @return destroy result
   */
-  public static Boolean shutdownWithoutReconstruction(ActorHandle actor) {
+  public static Boolean shutdownWithoutReconstruction(BaseActorHandle actor) {
    LOG.info("Call worker to shutdown without reconstruction, actor is {}.",
        actor.getId());
    Boolean result = false;
@@ -1,92 +0,0 @@
-package io.ray.streaming.runtime.schedule;
-
-import io.ray.api.ActorHandle;
-import io.ray.api.BaseActorHandle;
-import io.ray.api.ObjectRef;
-import io.ray.api.PyActorHandle;
-import io.ray.api.Ray;
-import io.ray.api.function.PyActorMethod;
-import io.ray.streaming.api.Language;
-import io.ray.streaming.jobgraph.JobGraph;
-import io.ray.streaming.runtime.core.graph.ExecutionGraph;
-import io.ray.streaming.runtime.core.graph.ExecutionNode;
-import io.ray.streaming.runtime.core.graph.ExecutionTask;
-import io.ray.streaming.runtime.generated.RemoteCall;
-import io.ray.streaming.runtime.python.GraphPbBuilder;
-import io.ray.streaming.runtime.worker.JobWorker;
-import io.ray.streaming.runtime.worker.context.WorkerContext;
-import io.ray.streaming.schedule.JobScheduler;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-/**
- * JobSchedulerImpl schedules workers by the Plan and the resource information
- * from ResourceManager.
- */
-public class JobSchedulerImpl implements JobScheduler {
-  private JobGraph jobGraph;
-  private Map<String, String> jobConfig;
-  private TaskAssigner taskAssigner;
-
-  public JobSchedulerImpl() {
-    this.taskAssigner = new TaskAssignerImpl();
-  }
-
-  /**
-   * Schedule physical plan to execution graph, and call streaming worker to init and run.
-   */
-  @SuppressWarnings("unchecked")
-  @Override
-  public void schedule(JobGraph jobGraph, Map<String, String> jobConfig) {
-    this.jobConfig = jobConfig;
-    this.jobGraph = jobGraph;
-
-    ExecutionGraph executionGraph = this.taskAssigner.assign(this.jobGraph);
-    List<ExecutionNode> executionNodes = executionGraph.getExecutionNodeList();
-    boolean hasPythonNode = executionNodes.stream()
-        .anyMatch(node -> node.getLanguage() == Language.PYTHON);
-    RemoteCall.ExecutionGraph executionGraphPb = null;
-    if (hasPythonNode) {
-      executionGraphPb = new GraphPbBuilder().buildExecutionGraphPb(executionGraph);
-    }
-    List<ObjectRef<Object>> waits = new ArrayList<>();
-    for (ExecutionNode executionNode : executionNodes) {
-      List<ExecutionTask> executionTasks = executionNode.getExecutionTasks();
-      for (ExecutionTask executionTask : executionTasks) {
-        int taskId = executionTask.getTaskId();
-        BaseActorHandle worker = executionTask.getWorker();
-        switch (executionNode.getLanguage()) {
-          case JAVA:
-            ActorHandle<JobWorker> jobWorker = (ActorHandle<JobWorker>) worker;
-            waits.add(jobWorker.call(JobWorker::init,
-                new WorkerContext(taskId, executionGraph, jobConfig)));
-            break;
-          case PYTHON:
-            byte[] workerContextBytes = buildPythonWorkerContext(
-                taskId, executionGraphPb, jobConfig);
-            waits.add(((PyActorHandle)worker).call(new PyActorMethod("init", Object.class),
-                workerContextBytes));
-            break;
-          default:
-            throw new UnsupportedOperationException(
-                "Unsupported language " + executionNode.getLanguage());
-        }
-      }
-    }
-    Ray.wait(waits);
-  }
-
-  private byte[] buildPythonWorkerContext(
-      int taskId,
-      RemoteCall.ExecutionGraph executionGraphPb,
-      Map<String, String> jobConfig) {
-    return RemoteCall.WorkerContext.newBuilder()
-        .setTaskId(taskId)
-        .putAllConf(jobConfig)
-        .setGraph(executionGraphPb)
-        .build()
-        .toByteArray();
-  }
-
-}
@@ -1,17 +0,0 @@
-package io.ray.streaming.runtime.schedule;
-
-import io.ray.streaming.jobgraph.JobGraph;
-import io.ray.streaming.runtime.core.graph.ExecutionGraph;
-import java.io.Serializable;
-
-/**
- * Interface of the task assigning strategy.
- */
-public interface TaskAssigner extends Serializable {
-
-  /**
-   * Assign logical plan to physical execution graph.
-   */
-  ExecutionGraph assign(JobGraph jobGraph);
-
-}
@@ -1,86 +0,0 @@
-package io.ray.streaming.runtime.schedule;
-
-import io.ray.api.ActorHandle;
-import io.ray.api.BaseActorHandle;
-import io.ray.api.PyActorHandle;
-import io.ray.api.Ray;
-import io.ray.api.function.PyActorClass;
-import io.ray.streaming.jobgraph.JobEdge;
-import io.ray.streaming.jobgraph.JobGraph;
-import io.ray.streaming.jobgraph.JobVertex;
-import io.ray.streaming.runtime.core.graph.ExecutionEdge;
-import io.ray.streaming.runtime.core.graph.ExecutionGraph;
-import io.ray.streaming.runtime.core.graph.ExecutionNode;
-import io.ray.streaming.runtime.core.graph.ExecutionTask;
-import io.ray.streaming.runtime.worker.JobWorker;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class TaskAssignerImpl implements TaskAssigner {
-  private static final Logger LOG = LoggerFactory.getLogger(TaskAssignerImpl.class);
-
-  /**
-   * Assign an optimized logical plan to execution graph.
-   *
-   * @param jobGraph The logical plan.
-   * @return The physical execution graph.
-   */
-  @Override
-  public ExecutionGraph assign(JobGraph jobGraph) {
-    List<JobVertex> jobVertices = jobGraph.getJobVertexList();
-    List<JobEdge> jobEdges = jobGraph.getJobEdgeList();
-
-    int taskId = 0;
-    Map<Integer, ExecutionNode> idToExecutionNode = new HashMap<>();
-    for (JobVertex jobVertex : jobVertices) {
-      ExecutionNode executionNode = new ExecutionNode(jobVertex.getVertexId(),
-          jobVertex.getParallelism(), jobVertex.getConfig());
-      executionNode.setNodeType(jobVertex.getVertexType());
-      List<ExecutionTask> vertexTasks = new ArrayList<>();
-      for (int taskIndex = 0; taskIndex < jobVertex.getParallelism(); taskIndex++) {
-        vertexTasks.add(new ExecutionTask(taskId, taskIndex, createWorker(jobVertex)));
-        taskId++;
-      }
-      executionNode.setExecutionTasks(vertexTasks);
-      executionNode.setStreamOperator(jobVertex.getStreamOperator());
-      idToExecutionNode.put(executionNode.getNodeId(), executionNode);
-    }
-
-    for (JobEdge jobEdge : jobEdges) {
-      int srcNodeId = jobEdge.getSrcVertexId();
-      int targetNodeId = jobEdge.getTargetVertexId();
-
-      ExecutionEdge executionEdge = new ExecutionEdge(srcNodeId, targetNodeId,
-          jobEdge.getPartition());
-      idToExecutionNode.get(srcNodeId).addOutputEdge(executionEdge);
-      idToExecutionNode.get(targetNodeId).addInputEdge(executionEdge);
-    }
-
-    List<ExecutionNode> executionNodes = new ArrayList<>(idToExecutionNode.values());
-    return new ExecutionGraph(executionNodes);
-  }
-
-  private BaseActorHandle createWorker(JobVertex jobVertex) {
-    switch (jobVertex.getLanguage()) {
-      case PYTHON: {
-        PyActorHandle worker = Ray.createActor(
-            new PyActorClass("ray.streaming.runtime.worker", "JobWorker"));
-        LOG.info("Created python worker {}", worker);
-        return worker;
-      }
-      case JAVA: {
-        ActorHandle<JobWorker> worker = Ray.createActor(JobWorker::new);
-        LOG.info("Created java worker {}", worker);
-        return worker;
-      }
-      default:
-        throw new UnsupportedOperationException(
-            "Unsupported language " + jobVertex.getLanguage());
-
-    }
-  }
-}
@@ -1,50 +1,68 @@
 package io.ray.streaming.runtime.transfer;

+import io.ray.streaming.runtime.config.StreamingWorkerConfig;
 import io.ray.streaming.runtime.generated.Streaming;
-import io.ray.streaming.util.Config;
-import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 public class ChannelUtils {
  private static final Logger LOGGER = LoggerFactory.getLogger(ChannelUtils.class);

-  static byte[] toNativeConf(Map<String, String> conf) {
+  static byte[] toNativeConf(StreamingWorkerConfig workerConfig) {
    Streaming.StreamingConfig.Builder builder = Streaming.StreamingConfig.newBuilder();
-    if (conf.containsKey(Config.STREAMING_JOB_NAME)) {
-      builder.setJobName(conf.get(Config.STREAMING_JOB_NAME));
+
+    // job name
+    String jobName = workerConfig.commonConfig.jobName();
+    if (!StringUtils.isEmpty(jobName)) {
+      builder.setJobName(workerConfig.commonConfig.jobName());
    }
-    if (conf.containsKey(Config.STREAMING_WORKER_NAME)) {
-      builder.setWorkerName(conf.get(Config.STREAMING_WORKER_NAME));
+
+    // worker name
+    String workerName = workerConfig.workerInternalConfig.workerName();
+    if (!StringUtils.isEmpty(workerName)) {
+      builder.setWorkerName(workerName);
    }
-    if (conf.containsKey(Config.STREAMING_OP_NAME)) {
-      builder.setOpName(conf.get(Config.STREAMING_OP_NAME));
+
+    // operator name
+    String operatorName = workerConfig.workerInternalConfig.workerOperatorName();
+    if (!StringUtils.isEmpty(operatorName)) {
+      builder.setOpName(operatorName);
    }
-    if (conf.containsKey(Config.STREAMING_RING_BUFFER_CAPACITY)) {
-      builder.setRingBufferCapacity(
-          Integer.parseInt(conf.get(Config.STREAMING_RING_BUFFER_CAPACITY)));
+
+    // ring buffer capacity
+    int ringBufferCapacity = workerConfig.transferConfig.ringBufferCapacity();
+    if (ringBufferCapacity != -1) {
+      builder.setRingBufferCapacity(ringBufferCapacity);
    }
-    if (conf.containsKey(Config.STREAMING_EMPTY_MESSAGE_INTERVAL)) {
-      builder.setEmptyMessageInterval(
-          Integer.parseInt(conf.get(Config.STREAMING_EMPTY_MESSAGE_INTERVAL)));
+
+    // empty message interval
+    int emptyMsgInterval = workerConfig.transferConfig.emptyMsgInterval();
+    if (emptyMsgInterval != -1) {
+      builder.setEmptyMessageInterval(emptyMsgInterval);
    }
-    if (conf.containsKey(Config.FLOW_CONTROL_TYPE)) {
-      builder.setFlowControlType(
-          Streaming.FlowControlType.forNumber(
-              Integer.parseInt(conf.get(Config.FLOW_CONTROL_TYPE))));
+
+    //flow control type
+    int flowControlType = workerConfig.transferConfig.flowControlType();
+    if (flowControlType != -1) {
+      builder.setFlowControlType(Streaming.FlowControlType.forNumber(flowControlType));
    }
-    if (conf.containsKey(Config.WRITER_CONSUMED_STEP)) {
-      builder.setWriterConsumedStep(
-          Integer.parseInt(conf.get(Config.WRITER_CONSUMED_STEP)));
+
+    // writer consumed step
+    int writerConsumedStep = workerConfig.transferConfig.writerConsumedStep();
+    if (writerConsumedStep != -1) {
+      builder.setWriterConsumedStep(writerConsumedStep);
    }
-    if (conf.containsKey(Config.READER_CONSUMED_STEP)) {
-      builder.setReaderConsumedStep(
-          Integer.parseInt(conf.get(Config.READER_CONSUMED_STEP)));
+
+    //reader consumed step
+    int readerConsumedStep = workerConfig.transferConfig.readerConsumedStep();
+    if (readerConsumedStep != -1) {
+      builder.setReaderConsumedStep(readerConsumedStep);
    }
+
    Streaming.StreamingConfig streamingConf = builder.build();
    LOGGER.info("Streaming native conf {}", streamingConf.toString());
    return streamingConf.toByteArray();
  }

 }
-
@@ -2,8 +2,9 @@ package io.ray.streaming.runtime.transfer;

 import com.google.common.base.Preconditions;
 import io.ray.api.BaseActorHandle;
+import io.ray.streaming.runtime.config.StreamingWorkerConfig;
+import io.ray.streaming.runtime.config.types.TransferChannelType;
 import io.ray.streaming.runtime.util.Platform;
-import io.ray.streaming.util.Config;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.util.LinkedList;
@@ -18,14 +19,19 @@ import org.slf4j.LoggerFactory;
 * from channels of upstream workers
 */
 public class DataReader {
-  private static final Logger LOGGER = LoggerFactory.getLogger(DataReader.class);
+  private static final Logger LOG = LoggerFactory.getLogger(DataReader.class);

  private long nativeReaderPtr;
  private Queue<DataMessage> buf = new LinkedList<>();

+  /**
+   * @param inputChannels input channels ids
+   * @param fromActors    upstream input actors
+   * @param workerConfig  configuration
+   */
  public DataReader(List<String> inputChannels,
                    Map<String, BaseActorHandle> fromActors,
-                    Map<String, String> conf) {
+                    StreamingWorkerConfig workerConfig) {
    Preconditions.checkArgument(inputChannels.size() > 0);
    Preconditions.checkArgument(inputChannels.size() == fromActors.size());
    ChannelCreationParametersBuilder initialParameters =
@@ -38,15 +44,14 @@ public class DataReader {
      seqIds[i] = 0;
      msgIds[i] = 0;
    }
-    long timerInterval = Long.parseLong(
-        conf.getOrDefault(Config.TIMER_INTERVAL_MS, "-1"));
-    String channelType = conf.get(Config.CHANNEL_TYPE);
+    long timerInterval = workerConfig.transferConfig.readerTimerIntervalMs();
+    TransferChannelType channelType = workerConfig.transferConfig.channelType();
    boolean isMock = false;
-    if (Config.MEMORY_CHANNEL.equals(channelType)) {
+    if (TransferChannelType.MEMORY_CHANNEL == channelType) {
      isMock = true;
    }
-    boolean isRecreate = Boolean.parseBoolean(
-        conf.getOrDefault(Config.IS_RECREATE, "false"));
+    boolean isRecreate = workerConfig.transferConfig.readerIsRecreate();
+
    this.nativeReaderPtr = createDataReaderNative(
        initialParameters,
        inputChannelsBytes,
@@ -54,10 +59,11 @@ public class DataReader {
        msgIds,
        timerInterval,
        isRecreate,
-        ChannelUtils.toNativeConf(conf),
+        ChannelUtils.toNativeConf(workerConfig),
        isMock
    );
-    LOGGER.info("create DataReader succeed");
+    LOG.info("Create DataReader succeed for worker: {}.",
+        workerConfig.workerInternalConfig.workerName());
  }

  // params set by getBundleNative: bundle data address + size
@@ -148,10 +154,10 @@ public class DataReader {
    if (nativeReaderPtr == 0) {
      return;
    }
-    LOGGER.info("closing DataReader.");
+    LOG.info("Closing DataReader.");
    closeReaderNative(nativeReaderPtr);
    nativeReaderPtr = 0;
-    LOGGER.info("closing DataReader done.");
+    LOG.info("Finish closing DataReader.");
  }

  private static native long createDataReaderNative(
@@ -2,8 +2,9 @@ package io.ray.streaming.runtime.transfer;

 import com.google.common.base.Preconditions;
 import io.ray.api.BaseActorHandle;
+import io.ray.streaming.runtime.config.StreamingWorkerConfig;
+import io.ray.streaming.runtime.config.types.TransferChannelType;
 import io.ray.streaming.runtime.util.Platform;
-import io.ray.streaming.util.Config;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.util.List;
@@ -17,7 +18,7 @@ import org.slf4j.LoggerFactory;
 * to downstream workers
 */
 public class DataWriter {
-  private static final Logger LOGGER = LoggerFactory.getLogger(DataWriter.class);
+  private static final Logger LOG = LoggerFactory.getLogger(DataWriter.class);

  private long nativeWriterPtr;
  private ByteBuffer buffer = ByteBuffer.allocateDirect(0);
@@ -30,38 +31,37 @@ public class DataWriter {
  /**
   * @param outputChannels output channels ids
   * @param toActors       downstream output actors
-   * @param conf           configuration
+   * @param workerConfig   configuration
   */
  public DataWriter(List<String> outputChannels,
                    Map<String, BaseActorHandle> toActors,
-                    Map<String, String> conf) {
+                    StreamingWorkerConfig workerConfig) {
    Preconditions.checkArgument(!outputChannels.isEmpty());
    Preconditions.checkArgument(outputChannels.size() == toActors.size());
-    ChannelCreationParametersBuilder initParameters =
+    ChannelCreationParametersBuilder initialParameters =
        new ChannelCreationParametersBuilder().buildOutputQueueParameters(outputChannels, toActors);
    byte[][] outputChannelsBytes = outputChannels.stream()
        .map(ChannelID::idStrToBytes).toArray(byte[][]::new);
-    long channelSize = Long.parseLong(
-        conf.getOrDefault(Config.CHANNEL_SIZE, Config.CHANNEL_SIZE_DEFAULT));
+    long channelSize = workerConfig.transferConfig.channelSize();
    long[] msgIds = new long[outputChannels.size()];
    for (int i = 0; i < outputChannels.size(); i++) {
      msgIds[i] = 0;
    }
-    String channelType = conf.get(Config.CHANNEL_TYPE);
+    TransferChannelType channelType = workerConfig.transferConfig.channelType();
    boolean isMock = false;
-    if (Config.MEMORY_CHANNEL.equalsIgnoreCase(channelType)) {
+    if (TransferChannelType.MEMORY_CHANNEL == channelType) {
      isMock = true;
-      LOGGER.info("Using memory channel");
    }
    this.nativeWriterPtr = createWriterNative(
-        initParameters,
+        initialParameters,
        outputChannelsBytes,
        msgIds,
        channelSize,
-        ChannelUtils.toNativeConf(conf),
+        ChannelUtils.toNativeConf(workerConfig),
        isMock
    );
-    LOGGER.info("create DataWriter succeed");
+    LOG.info("Create DataWriter succeed for worker: {}.",
+        workerConfig.workerInternalConfig.workerName());
  }

  /**
@@ -117,10 +117,10 @@ public class DataWriter {
    if (nativeWriterPtr == 0) {
      return;
    }
-    LOGGER.info("closing data writer.");
+    LOG.info("Closing data writer.");
    closeWriterNative(nativeWriterPtr);
    nativeWriterPtr = 0;
-    LOGGER.info("closing data writer done.");
+    LOG.info("Finish closing data writer.");
  }

  private static native long createWriterNative(
@@ -0,0 +1,14 @@
+package io.ray.streaming.runtime.util;
+
+import java.util.Map;
+
+/**
+ * Common tools.
+ */
+public class CommonUtils {
+
+  public static Map<String, Object> strMapToObjectMap(Map<String, String> srcMap) {
+    Map<String,Object> destMap = (Map) srcMap;
+    return destMap;
+  }
+}
@@ -1,30 +1,35 @@
 package io.ray.streaming.runtime.worker;

-import io.ray.api.Ray;
-import io.ray.streaming.runtime.core.graph.ExecutionGraph;
-import io.ray.streaming.runtime.core.graph.ExecutionNode;
-import io.ray.streaming.runtime.core.graph.ExecutionNode.NodeType;
-import io.ray.streaming.runtime.core.graph.ExecutionTask;
+import io.ray.streaming.runtime.config.StreamingWorkerConfig;
+import io.ray.streaming.runtime.config.types.TransferChannelType;
+import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionVertex;
 import io.ray.streaming.runtime.core.processor.OneInputProcessor;
 import io.ray.streaming.runtime.core.processor.ProcessBuilder;
 import io.ray.streaming.runtime.core.processor.SourceProcessor;
 import io.ray.streaming.runtime.core.processor.StreamProcessor;
+import io.ray.streaming.runtime.master.JobMaster;
 import io.ray.streaming.runtime.transfer.TransferHandler;
 import io.ray.streaming.runtime.util.EnvUtil;
-import io.ray.streaming.runtime.worker.context.WorkerContext;
+import io.ray.streaming.runtime.worker.context.JobWorkerContext;
 import io.ray.streaming.runtime.worker.tasks.OneInputStreamTask;
 import io.ray.streaming.runtime.worker.tasks.SourceStreamTask;
 import io.ray.streaming.runtime.worker.tasks.StreamTask;
 import java.io.Serializable;
-import java.util.Map;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /**
- * The stream job worker, it is a ray actor.
+ * The streaming worker implementation class, it is ray actor. JobWorker is created by
+ * {@link JobMaster} through ray api, and JobMaster communicates
+ * with JobWorker through Ray.call().
+ *
+ * <p>The JobWorker is responsible for creating tasks and defines the methods of communication
+ * between workers.
 */
 public class JobWorker implements Serializable {
-  private static final Logger LOGGER = LoggerFactory.getLogger(JobWorker.class);
+
+  private static final Logger LOG = LoggerFactory.getLogger(JobWorker.class);
+
  // special flag to indicate this actor not ready
  private static final byte[] NOT_READY_FLAG = new byte[4];

@@ -32,79 +37,102 @@ public class JobWorker implements Serializable {
    EnvUtil.loadNativeLibraries();
  }

-  private int taskId;
-  private Map<String, String> config;
-  private WorkerContext workerContext;
-  private ExecutionNode executionNode;
-  private ExecutionTask executionTask;
-  private ExecutionGraph executionGraph;
-  private StreamProcessor streamProcessor;
-  private NodeType nodeType;
+  private JobWorkerContext workerContext;
+  private ExecutionVertex executionVertex;
+  private StreamingWorkerConfig workerConfig;
+
  private StreamTask task;
  private TransferHandler transferHandler;

-  public Boolean init(WorkerContext workerContext) {
-    this.workerContext = workerContext;
-    this.taskId = workerContext.getTaskId();
-    this.config = workerContext.getConfig();
-    this.executionGraph = this.workerContext.getExecutionGraph();
-    this.executionTask = executionGraph.getExecutionTaskByTaskId(taskId);
-    this.executionNode = executionGraph.getExecutionNodeByTaskId(taskId);
+  public JobWorker() {
+    LOG.info("Creating job worker succeeded.");
+  }

-    this.nodeType = executionNode.getNodeType();
-    this.streamProcessor = ProcessBuilder
-        .buildProcessor(executionNode.getStreamOperator());
-    LOGGER.info("Initializing StreamWorker, pid {}, taskId: {}, operator: {}.",
-        EnvUtil.getJvmPid(), taskId, streamProcessor);
+  /**
+   * Initialize JobWorker and data communication pipeline.
+   */
+  public Boolean init(JobWorkerContext workerContext) {
+    LOG.info("Initiating job worker: {}. Worker context is: {}.",
+        workerContext.getWorkerName(), workerContext);

-    if (!Ray.getRuntimeContext().isSingleProcess()) {
-      transferHandler = new TransferHandler();
+    try {
+      this.workerContext = workerContext;
+      this.executionVertex = workerContext.getExecutionVertex();
+      this.workerConfig = new StreamingWorkerConfig(executionVertex.getWorkerConfig());
+
+      //Init transfer
+      TransferChannelType channelType = workerConfig.transferConfig.channelType();
+      if (TransferChannelType.NATIVE_CHANNEL == channelType) {
+        transferHandler = new TransferHandler();
+      }
+
+      // create stream task
+      task = createStreamTask();
+      if (task == null) {
+        return false;
+      }
+    } catch (Exception e) {
+      LOG.error("Failed to initiate job worker.", e);
+      return false;
    }
-    task = createStreamTask();
-    task.start();
+    LOG.info("Initiating job worker succeeded: {}.", workerContext.getWorkerName());
    return true;
  }

-  private StreamTask createStreamTask() {
-    if (streamProcessor instanceof OneInputProcessor) {
-      return new OneInputStreamTask(taskId, streamProcessor, this);
-    } else if (streamProcessor instanceof SourceProcessor) {
-      return new SourceStreamTask(taskId, streamProcessor, this);
-    } else {
-      throw new RuntimeException("Unsupported type: " + streamProcessor);
+  /**
+   * Start worker's stream tasks.
+   *
+   * @return result
+   */
+  public Boolean start() {
+    try {
+      task.start();
+    } catch (Exception e) {
+      LOG.error("Start worker [{}] occur error.", executionVertex.getExecutionVertexName(), e);
+      return false;
    }
+    return true;
+  }
+
+  /**
+   * Create tasks based on the processor corresponding of the operator.
+   */
+  private StreamTask createStreamTask() {
+    StreamTask task = null;
+    StreamProcessor streamProcessor = ProcessBuilder
+        .buildProcessor(executionVertex.getStreamOperator());
+    LOG.debug("Stream processor created: {}.", streamProcessor);
+
+    try {
+      if (streamProcessor instanceof SourceProcessor) {
+        task = new SourceStreamTask(getTaskId(), streamProcessor, this);
+      } else if (streamProcessor instanceof OneInputProcessor) {
+        task = new OneInputStreamTask(getTaskId(), streamProcessor, this);
+      } else {
+        throw new RuntimeException("Unsupported processor type:" + streamProcessor);
+      }
+    } catch (Exception e) {
+      LOG.info("Failed to create stream task.", e);
+      return task;
+    }
+    LOG.info("Stream task created: {}.", task);
+    return task;
  }

  public int getTaskId() {
-    return taskId;
+    return executionVertex.getExecutionVertexId();
  }

-  public Map<String, String> getConfig() {
-    return config;
+  public StreamingWorkerConfig getWorkerConfig() {
+    return workerConfig;
  }

-  public WorkerContext getWorkerContext() {
+  public JobWorkerContext getWorkerContext() {
    return workerContext;
  }

-  public NodeType getNodeType() {
-    return nodeType;
-  }
-
-  public ExecutionNode getExecutionNode() {
-    return executionNode;
-  }
-
-  public ExecutionTask getExecutionTask() {
-    return executionTask;
-  }
-
-  public ExecutionGraph getExecutionGraph() {
-    return executionGraph;
-  }
-
-  public StreamProcessor getStreamProcessor() {
-    return streamProcessor;
+  public ExecutionVertex getExecutionVertex() {
+    return executionVertex;
  }

  public StreamTask getTask() {
@@ -1,22 +1,21 @@
 package io.ray.streaming.runtime.worker.context;

 import com.google.common.base.MoreObjects;
+import com.google.protobuf.ByteString;
 import io.ray.api.ActorHandle;
-import io.ray.api.id.ActorId;
+import io.ray.runtime.actor.NativeActorHandle;
 import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionVertex;
+import io.ray.streaming.runtime.generated.RemoteCall;
 import io.ray.streaming.runtime.master.JobMaster;
+import io.ray.streaming.runtime.python.GraphPbBuilder;
 import java.io.Serializable;
+import java.util.Map;

 /**
- * Job worker context.
+ * Job worker context of java type.
 */
 public class JobWorkerContext implements Serializable {

-  /**
-   * Worker actor's id.
-   */
-  private ActorId workerId;
-
  /**
   * JobMaster actor.
   */
@@ -28,16 +27,22 @@ public class JobWorkerContext implements Serializable {
  private ExecutionVertex executionVertex;

  public JobWorkerContext(
-      ActorId workerId,
      ActorHandle<JobMaster> master,
      ExecutionVertex executionVertex) {
-    this.workerId = workerId;
    this.master = master;
    this.executionVertex = executionVertex;
  }

-  public ActorId getWorkerId() {
-    return workerId;
+  public int getWorkerId() {
+    return executionVertex.getExecutionVertexId();
+  }
+
+  public String getWorkerName() {
+    return executionVertex.getExecutionVertexName();
+  }
+
+  public Map<String, String> getConfig() {
+    return executionVertex.getWorkerConfig();
  }

  public ActorHandle<JobMaster> getMaster() {
@@ -51,9 +56,24 @@ public class JobWorkerContext implements Serializable {
  @Override
  public String toString() {
    return MoreObjects.toStringHelper(this)
-      .add("workerId", workerId)
-      .add("master", master)
+      .add("workerId", getWorkerId())
+      .add("workerName", getWorkerName())
+      .add("config", getConfig())
      .toString();
  }

+  public byte[] getPythonWorkerContextBytes() {
+    // create python worker context
+    RemoteCall.ExecutionVertexContext executionVertexContext =
+        new GraphPbBuilder().buildExecutionVertexContext(executionVertex);
+
+    byte[] contextBytes = RemoteCall.PythonJobWorkerContext.newBuilder()
+      .setMasterActor(ByteString.copyFrom((((NativeActorHandle) (master)).toBytes())))
+      .setExecutionVertexContext(executionVertexContext)
+      .build()
+      .toByteArray();
+
+    return contextBytes;
+  }
+
 }
@@ -2,7 +2,7 @@ package io.ray.streaming.runtime.worker.context;

 import com.google.common.base.Preconditions;
 import io.ray.streaming.api.context.RuntimeContext;
-import io.ray.streaming.runtime.core.graph.ExecutionTask;
+import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionVertex;
 import io.ray.streaming.state.backend.AbstractKeyStateBackend;
 import io.ray.streaming.state.backend.KeyStateBackend;
 import io.ray.streaming.state.backend.OperatorStateBackend;
@@ -18,7 +18,7 @@ import java.util.Map;
 /**
 * Use Ray to implement RuntimeContext.
 */
-public class RayRuntimeContext implements RuntimeContext {
+public class StreamingRuntimeContext implements RuntimeContext {
  /**
   * Backend for keyed state. This might be empty if we're not on a keyed stream.
   */
@@ -33,11 +33,11 @@ public class RayRuntimeContext implements RuntimeContext {
  private Long checkpointId;
  private Map<String, String> config;

-  public RayRuntimeContext(ExecutionTask executionTask, Map<String, String> config,
+  public StreamingRuntimeContext(ExecutionVertex executionVertex, Map<String, String> config,
      int parallelism) {
-    this.taskId = executionTask.getTaskId();
+    this.taskId = executionVertex.getExecutionVertexId();
    this.config = config;
-    this.taskIndex = executionTask.getTaskIndex();
+    this.taskIndex = executionVertex.getExecutionVertexIndex();
    this.parallelism = parallelism;
  }

@@ -1,41 +0,0 @@
-package io.ray.streaming.runtime.worker.context;
-
-import io.ray.streaming.runtime.core.graph.ExecutionGraph;
-import java.io.Serializable;
-import java.util.Map;
-
-/**
- * Encapsulate the context information for worker initialization.
- */
-public class WorkerContext implements Serializable {
-
-  private int taskId;
-  private ExecutionGraph executionGraph;
-  private Map<String, String> config;
-
-  public WorkerContext(int taskId, ExecutionGraph executionGraph, Map<String, String> jobConfig) {
-    this.taskId = taskId;
-    this.executionGraph = executionGraph;
-    this.config = jobConfig;
-  }
-
-  public int getTaskId() {
-    return taskId;
-  }
-
-  public void setTaskId(int taskId) {
-    this.taskId = taskId;
-  }
-
-  public ExecutionGraph getExecutionGraph() {
-    return executionGraph;
-  }
-
-  public void setExecutionGraph(ExecutionGraph executionGraph) {
-    this.executionGraph = executionGraph;
-  }
-
-  public Map<String, String> getConfig() {
-    return config;
-  }
-}
@@ -1,12 +1,12 @@
 package io.ray.streaming.runtime.worker.tasks;

+import com.google.common.base.MoreObjects;
 import io.ray.streaming.runtime.core.processor.Processor;
 import io.ray.streaming.runtime.serialization.CrossLangSerializer;
 import io.ray.streaming.runtime.serialization.JavaSerializer;
 import io.ray.streaming.runtime.serialization.Serializer;
 import io.ray.streaming.runtime.transfer.Message;
 import io.ray.streaming.runtime.worker.JobWorker;
-import io.ray.streaming.util.Config;

 public abstract class InputStreamTask extends StreamTask {
  private volatile boolean running = true;
@@ -15,10 +15,9 @@ public abstract class InputStreamTask extends StreamTask {
  private final io.ray.streaming.runtime.serialization.Serializer javaSerializer;
  private final io.ray.streaming.runtime.serialization.Serializer crossLangSerializer;

-  public InputStreamTask(int taskId, Processor processor, JobWorker streamWorker) {
-    super(taskId, processor, streamWorker);
-    readTimeoutMillis = Long.parseLong((String) streamWorker.getConfig()
-        .getOrDefault(Config.READ_TIMEOUT_MS, Config.DEFAULT_READ_TIMEOUT_MS));
+  public InputStreamTask(int taskId, Processor processor, JobWorker jobWorker) {
+    super(taskId, processor, jobWorker);
+    readTimeoutMillis = jobWorker.getWorkerConfig().transferConfig.readerTimerIntervalMs();
    javaSerializer = new JavaSerializer();
    crossLangSerializer = new CrossLangSerializer();
  }
@@ -56,10 +55,9 @@ public abstract class InputStreamTask extends StreamTask {

  @Override
  public String toString() {
-    final StringBuilder sb = new StringBuilder("InputStreamTask{");
-    sb.append("taskId=").append(taskId);
-    sb.append(", processor=").append(processor);
-    sb.append('}');
-    return sb.toString();
+    return MoreObjects.toStringHelper(this)
+      .add("taskId", taskId)
+      .add("processor", processor)
+      .toString();
  }
 }
@@ -3,9 +3,12 @@ package io.ray.streaming.runtime.worker.tasks;
 import io.ray.streaming.runtime.core.processor.Processor;
 import io.ray.streaming.runtime.worker.JobWorker;

-public class OneInputStreamTask<IN> extends InputStreamTask {
+/**
+ * Input stream task with 1 input. Such as: map operator.
+ */
+public class OneInputStreamTask extends InputStreamTask {

-  public OneInputStreamTask(int taskId, Processor processor, JobWorker streamWorker) {
-    super(taskId, processor, streamWorker);
+  public OneInputStreamTask(int taskId, Processor inputProcessor, JobWorker jobWorker) {
+    super(taskId, inputProcessor, jobWorker);
  }
 }
@@ -1,16 +1,25 @@
 package io.ray.streaming.runtime.worker.tasks;

+import io.ray.streaming.operator.impl.SourceOperator;
 import io.ray.streaming.runtime.core.processor.Processor;
 import io.ray.streaming.runtime.core.processor.SourceProcessor;
 import io.ray.streaming.runtime.worker.JobWorker;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

-public class SourceStreamTask<IN> extends StreamTask {
-  private static final Logger LOGGER = LoggerFactory.getLogger(SourceStreamTask.class);
+public class SourceStreamTask extends StreamTask {

-  public SourceStreamTask(int taskId, Processor processor, JobWorker worker) {
-    super(taskId, processor, worker);
+  private static final Logger LOG = LoggerFactory.getLogger(SourceStreamTask.class);
+
+  private final SourceProcessor sourceProcessor;
+
+  /**
+   * SourceStreamTask for executing a {@link SourceOperator}.
+   * It is responsible for running the corresponding source operator.
+   */
+  public SourceStreamTask(int taskId, Processor sourceProcessor, JobWorker jobWorker) {
+    super(taskId, sourceProcessor, jobWorker);
+    this.sourceProcessor = (SourceProcessor) processor;
  }

  @Override
@@ -19,12 +28,15 @@ public class SourceStreamTask<IN> extends StreamTask {

  @Override
  public void run() {
-    final SourceProcessor<IN> sourceProcessor = (SourceProcessor<IN>) this.processor;
-    sourceProcessor.run();
+    LOG.info("Source stream task thread start.");
+
+    while (running) {
+      sourceProcessor.run();
+    }
  }

  @Override
-  protected void cancelTask() throws Exception {
+  protected void cancelTask() {
  }

 }
@@ -4,24 +4,20 @@ import io.ray.api.BaseActorHandle;
 import io.ray.api.Ray;
 import io.ray.streaming.api.collector.Collector;
 import io.ray.streaming.api.context.RuntimeContext;
-import io.ray.streaming.api.partition.Partition;
+import io.ray.streaming.runtime.config.worker.WorkerInternalConfig;
 import io.ray.streaming.runtime.core.collector.OutputCollector;
-import io.ray.streaming.runtime.core.graph.ExecutionEdge;
-import io.ray.streaming.runtime.core.graph.ExecutionGraph;
-import io.ray.streaming.runtime.core.graph.ExecutionNode;
+import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionEdge;
+import io.ray.streaming.runtime.core.graph.executiongraph.ExecutionVertex;
 import io.ray.streaming.runtime.core.processor.Processor;
 import io.ray.streaming.runtime.transfer.ChannelID;
 import io.ray.streaming.runtime.transfer.DataReader;
 import io.ray.streaming.runtime.transfer.DataWriter;
 import io.ray.streaming.runtime.worker.JobWorker;
-import io.ray.streaming.runtime.worker.context.RayRuntimeContext;
-import io.ray.streaming.util.Config;
-
+import io.ray.streaming.runtime.worker.context.StreamingRuntimeContext;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

@@ -30,71 +26,71 @@ public abstract class StreamTask implements Runnable {

  protected int taskId;
  protected Processor processor;
-  protected JobWorker worker;
+  protected JobWorker jobWorker;
  protected DataReader reader;
-  private Map<ExecutionEdge, DataWriter> writers;
+  List<Collector> collectors = new ArrayList<>();
+
+  protected volatile boolean running = true;
+  protected volatile boolean stopped = false;
+
  private Thread thread;

-  public StreamTask(int taskId, Processor processor, JobWorker worker) {
+  protected StreamTask(int taskId, Processor processor, JobWorker jobWorker) {
    this.taskId = taskId;
    this.processor = processor;
-    this.worker = worker;
+    this.jobWorker = jobWorker;
    prepareTask();

    this.thread = new Thread(Ray.wrapRunnable(this),
-        this.getClass().getName() + "-" + System.currentTimeMillis());
+      this.getClass().getName() + "-" + System.currentTimeMillis());
    this.thread.setDaemon(true);
  }

+  /**
+   * Build upstream and downstream data transmission channels according to {@link ExecutionVertex}.
+   */
  private void prepareTask() {
-    Map<String, String> queueConf = new HashMap<>();
-    worker.getConfig().forEach((k, v) -> queueConf.put(k, String.valueOf(v)));
-    String queueSize = worker.getConfig()
-        .getOrDefault(Config.CHANNEL_SIZE, Config.CHANNEL_SIZE_DEFAULT);
-    queueConf.put(Config.CHANNEL_SIZE, queueSize);
-    String channelType = Ray.getRuntimeContext().isSingleProcess() ?
-        Config.MEMORY_CHANNEL : Config.NATIVE_CHANNEL;
-    queueConf.put(Config.CHANNEL_TYPE, channelType);
+    LOG.debug("Preparing stream task.");
+    ExecutionVertex executionVertex = jobWorker.getExecutionVertex();

-    ExecutionGraph executionGraph = worker.getExecutionGraph();
-    ExecutionNode executionNode = worker.getExecutionNode();
+    // set vertex info into config for native using
+    jobWorker.getWorkerConfig().workerInternalConfig.setProperty(
+        WorkerInternalConfig.WORKER_NAME_INTERNAL, executionVertex.getExecutionVertexName());
+    jobWorker.getWorkerConfig().workerInternalConfig.setProperty(
+        WorkerInternalConfig.OP_NAME_INTERNAL, executionVertex.getExecutionJobVertexName());
+
+    // producer
+    List<ExecutionEdge> outputEdges = executionVertex.getOutputEdges();
+    Map<String, BaseActorHandle> outputActors = new HashMap<>();

-    // writers
-    writers = new HashMap<>();
-    List<ExecutionEdge> outputEdges = executionNode.getOutputEdges();
-    List<Collector> collectors = new ArrayList<>();
    for (ExecutionEdge edge : outputEdges) {
-      Map<String, BaseActorHandle> outputActors = new HashMap<>();
-      Map<Integer, BaseActorHandle> taskId2Worker = executionGraph
-          .getTaskId2WorkerByNodeId(edge.getTargetNodeId());
-      taskId2Worker.forEach((targetTaskId, targetActor) -> {
-        String queueName = ChannelID.genIdStr(taskId, targetTaskId, executionGraph.getBuildTime());
-        outputActors.put(queueName, targetActor);
+      String queueName = ChannelID.genIdStr(
+          taskId,
+          edge.getTargetExecutionVertex().getExecutionVertexId(),
+          executionVertex.getBuildTime());
+      outputActors.put(queueName, edge.getTargetExecutionVertex().getWorkerActor());
+    }
+
+    if (!outputActors.isEmpty()) {
+      List<String> channelIDs = new ArrayList<>();
+      outputActors.forEach((vertexId, actorId) -> {
+        channelIDs.add(vertexId);
      });

-      if (!outputActors.isEmpty()) {
-        List<String> channelIDs = new ArrayList<>();
-        outputActors.forEach((k, v) -> {
-          channelIDs.add(k);
-        });
-        DataWriter writer = new DataWriter(channelIDs, outputActors, queueConf);
-        LOG.info("Create DataWriter succeed.");
-        writers.put(edge, writer);
-        Partition partition = edge.getPartition();
-        collectors.add(new OutputCollector(writer, channelIDs, outputActors.values(), partition));
-      }
+      DataWriter writer = new DataWriter(channelIDs, outputActors, jobWorker.getWorkerConfig());
+      collectors.add(new OutputCollector(writer, channelIDs, outputActors.values(),
+          executionVertex.getOutputEdges().get(0).getPartition()));
    }

    // consumer
-    List<ExecutionEdge> inputEdges = executionNode.getInputsEdges();
+    List<ExecutionEdge> inputEdges = executionVertex.getInputEdges();
    Map<String, BaseActorHandle> inputActors = new HashMap<>();
    for (ExecutionEdge edge : inputEdges) {
-      Map<Integer, BaseActorHandle> taskId2Worker = executionGraph
-          .getTaskId2WorkerByNodeId(edge.getSrcNodeId());
-      taskId2Worker.forEach((srcTaskId, srcActor) -> {
-        String queueName = ChannelID.genIdStr(srcTaskId, taskId, executionGraph.getBuildTime());
-        inputActors.put(queueName, srcActor);
-      });
+      String queueName = ChannelID.genIdStr(
+          edge.getSourceExecutionVertex().getExecutionVertexId(),
+          taskId,
+          executionVertex.getBuildTime());
+      inputActors.put(queueName, edge.getSourceExecutionVertex().getWorkerActor());
    }
    if (!inputActors.isEmpty()) {
      List<String> channelIDs = new ArrayList<>();
@@ -102,31 +98,43 @@ public abstract class StreamTask implements Runnable {
        channelIDs.add(k);
      });
      LOG.info("Register queue consumer, queues {}.", channelIDs);
-      reader = new DataReader(channelIDs, inputActors, queueConf);
+      reader = new DataReader(channelIDs, inputActors, jobWorker.getWorkerConfig());
    }

-    RuntimeContext runtimeContext = new RayRuntimeContext(
-        worker.getExecutionTask(), worker.getConfig(), executionNode.getParallelism());
+    RuntimeContext runtimeContext = new StreamingRuntimeContext(executionVertex,
+        jobWorker.getWorkerConfig().configMap, executionVertex.getParallelism());

    processor.open(collectors, runtimeContext);
-
-    Runtime.getRuntime().addShutdownHook(new Thread(() -> {
-      try {
-        // Make DataReader stop read data when MockQueue destructor gets called to avoid crash
-        StreamTask.this.cancelTask();
-      } catch (Exception e) {
-        e.printStackTrace();
-      }
-    }));
+    LOG.debug("Finished preparing stream task.");
  }

+  /**
+   * Task initialization related work.
+   */
  protected abstract void init() throws Exception;

+  /**
+   * Stop running tasks.
+   */
  protected abstract void cancelTask() throws Exception;

  public void start() {
+    LOG.info("Start stream task: {}-{}", this.getClass().getSimpleName(), taskId);
    this.thread.start();
-    LOG.info("started {}-{}", this.getClass().getSimpleName(), taskId);
+  }
+
+  /**
+   * Close running tasks.
+   */
+  public void close() {
+    this.running = false;
+    if (thread.isAlive() && !Ray.getRuntimeContext().isSingleProcess()) {
+      // `Runtime.halt` is used because System.exist can't ensure the process killing.
+      Runtime.getRuntime().halt(0);
+      LOG.warn("runtime halt 0");
+      System.exit(0);
+    }
+    LOG.info("Stream task close success.");
  }

 }
@@ -0,0 +1,23 @@
+package io.ray.streaming.runtime.worker.tasks;
+
+import io.ray.streaming.runtime.core.processor.Processor;
+import io.ray.streaming.runtime.core.processor.TwoInputProcessor;
+import io.ray.streaming.runtime.worker.JobWorker;
+
+/**
+ * Input stream task with 2 inputs. Such as: join operator.
+ */
+public class TwoInputStreamTask extends InputStreamTask {
+
+  public TwoInputStreamTask(
+      int taskId,
+      Processor processor,
+      JobWorker jobWorker,
+      String leftStream,
+      String rightStream) {
+    super(taskId, processor, jobWorker);
+    ((TwoInputProcessor)(super.processor)).setLeftStream(leftStream);
+    ((TwoInputProcessor)(super.processor)).setRightStream(rightStream);
+  }
+
+}
@@ -0,0 +1 @@
+io.ray.streaming.runtime.client.JobClientImpl
@@ -1 +0,0 @@
-io.ray.streaming.runtime.schedule.JobSchedulerImpl
@@ -35,6 +35,8 @@ public class ExecutionGraphTest extends BaseUnitTest {
    StreamingConfig streamingConfig = new StreamingConfig(jobConf);
    GraphManager graphManager = new GraphManagerImpl(new JobRuntimeContext(streamingConfig));
    JobGraph jobGraph = buildJobGraph();
+    jobGraph.getJobConfig().put("streaming.task.resource.cpu.limitation.enable", "true");
+
    ExecutionGraph executionGraph = buildExecutionGraph(graphManager, jobGraph);
    List<ExecutionJobVertex> executionJobVertices = executionGraph.getExecutionJobVertexList();

@@ -48,24 +50,24 @@ public class ExecutionGraphTest extends BaseUnitTest {

    executionGraph.getAllExecutionVertices().forEach(vertex -> {
        Assert.assertNotNull(vertex.getStreamOperator());
-        Assert.assertNotNull(vertex.getJobVertexName());
+        Assert.assertNotNull(vertex.getExecutionJobVertexName());
        Assert.assertNotNull(vertex.getVertexType());
        Assert.assertNotNull(vertex.getLanguage());
-        Assert.assertEquals(vertex.getVertexName(),
-          vertex.getJobVertexId() + "-" + vertex.getJobVertexName() + "-" + vertex.getVertexIndex());
+        Assert.assertEquals(vertex.getExecutionVertexName(),
+          vertex.getExecutionJobVertexName() + "-" + vertex.getExecutionVertexIndex());
    });

    int startIndex = 0;
    ExecutionJobVertex upStream = executionJobVertices.get(startIndex);
    ExecutionJobVertex downStream = executionJobVertices.get(startIndex + 1);
-    Assert.assertEquals(upStream.getOutputEdges().get(0).getTargetVertex(), downStream);
+    Assert.assertEquals(upStream.getOutputEdges().get(0).getTargetExecutionJobVertex(), downStream);

    List<ExecutionVertex> upStreamVertices = upStream.getExecutionVertices();
    List<ExecutionVertex> downStreamVertices = downStream.getExecutionVertices();
    upStreamVertices.forEach(vertex -> {
-        Assert.assertEquals(vertex.getResources().get(ResourceType.CPU.name()), 2.0);
+        Assert.assertEquals(vertex.getResource().get(ResourceType.CPU.name()), 2.0);
        vertex.getOutputEdges().stream().forEach(upStreamOutPutEdge -> {
-            Assert.assertTrue(downStreamVertices.contains(upStreamOutPutEdge.getTargetVertex()));
+            Assert.assertTrue(downStreamVertices.contains(upStreamOutPutEdge.getTargetExecutionVertex()));
        });
    });
  }
@@ -21,18 +21,17 @@ import org.testng.annotations.Test;

 public class WordCountTest extends BaseUnitTest implements Serializable {

-  private static final Logger LOGGER = LoggerFactory.getLogger(WordCountTest.class);
+  private static final Logger LOG = LoggerFactory.getLogger(WordCountTest.class);

-  // TODO(zhenxuanpan): this test only works in single-process mode, because we put
-  //   results in this in-memory map.
  static Map<String, Integer> wordCount = new ConcurrentHashMap<>();

  @Test(timeOut = 60000)
  public void testWordCount() {
    Ray.shutdown();
+
    StreamingContext streamingContext = StreamingContext.buildContext();
    Map<String, String> config = new HashMap<>();
-    config.put(Config.CHANNEL_TYPE, Config.MEMORY_CHANNEL);
+    config.put(Config.CHANNEL_TYPE, "MEMORY_CHANNEL");
    streamingContext.withConfig(config);
    List<String> text = new ArrayList<>();
    text.add("hello world eagle eagle eagle");
@@ -58,7 +57,7 @@ public class WordCountTest extends BaseUnitTest implements Serializable {
      try {
        Thread.sleep(1000);
      } catch (InterruptedException e) {
-        LOGGER.warn("Got an exception while sleeping.", e);
+        LOG.warn("Got an exception while sleeping.", e);
      }
    }
    streamingContext.stop();
@@ -2,7 +2,7 @@ package io.ray.streaming.runtime.master.jobscheduler;

 import org.testng.annotations.Test;

-public class JobSchedulerTest {
+public class JobClientTest {

  @Test
  public void testSchedule() {
@@ -1,67 +0,0 @@
-package io.ray.streaming.runtime.schedule;
-
-import com.google.common.collect.Lists;
-import io.ray.api.Ray;
-import io.ray.streaming.api.context.StreamingContext;
-import io.ray.streaming.api.partition.impl.RoundRobinPartition;
-import io.ray.streaming.api.stream.DataStream;
-import io.ray.streaming.api.stream.DataStreamSink;
-import io.ray.streaming.api.stream.DataStreamSource;
-import io.ray.streaming.jobgraph.JobGraph;
-import io.ray.streaming.jobgraph.JobGraphBuilder;
-import io.ray.streaming.runtime.BaseUnitTest;
-import io.ray.streaming.runtime.core.graph.ExecutionEdge;
-import io.ray.streaming.runtime.core.graph.ExecutionGraph;
-import io.ray.streaming.runtime.core.graph.ExecutionNode;
-import io.ray.streaming.runtime.core.graph.ExecutionNode.NodeType;
-import java.util.List;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testng.Assert;
-import org.testng.annotations.Test;
-
-public class TaskAssignerImplTest extends BaseUnitTest {
-
-  private static final Logger LOGGER = LoggerFactory.getLogger(TaskAssignerImplTest.class);
-
-  @Test
-  public void testTaskAssignImpl() {
-    Ray.init();
-    JobGraph jobGraph = buildDataSyncPlan();
-
-    TaskAssigner taskAssigner = new TaskAssignerImpl();
-    ExecutionGraph executionGraph = taskAssigner.assign(jobGraph);
-
-    List<ExecutionNode> executionNodeList = executionGraph.getExecutionNodeList();
-
-    Assert.assertEquals(executionNodeList.size(), 2);
-    ExecutionNode sourceNode = executionNodeList.get(0);
-    Assert.assertEquals(sourceNode.getNodeType(), NodeType.SOURCE);
-    Assert.assertEquals(sourceNode.getExecutionTasks().size(), 1);
-    Assert.assertEquals(sourceNode.getOutputEdges().size(), 1);
-
-    List<ExecutionEdge> sourceExecutionEdges = sourceNode.getOutputEdges();
-
-    Assert.assertEquals(sourceExecutionEdges.size(), 1);
-    ExecutionEdge source2Sink = sourceExecutionEdges.get(0);
-
-    Assert.assertEquals(source2Sink.getPartition().getClass(), RoundRobinPartition.class);
-
-    ExecutionNode sinkNode = executionNodeList.get(1);
-    Assert.assertEquals(sinkNode.getNodeType(), NodeType.SINK);
-    Assert.assertEquals(sinkNode.getExecutionTasks().size(), 1);
-    Assert.assertEquals(sinkNode.getOutputEdges().size(), 0);
-
-    Ray.shutdown();
-  }
-
-  public JobGraph buildDataSyncPlan() {
-    StreamingContext streamingContext = StreamingContext.buildContext();
-    DataStream<String> dataStream = DataStreamSource.fromCollection(streamingContext,
-        Lists.newArrayList("a", "b", "c"));
-    DataStreamSink streamSink = dataStream.sink(LOGGER::info);
-    JobGraphBuilder jobGraphBuilder = new JobGraphBuilder(Lists.newArrayList(streamSink));
-
-    return jobGraphBuilder.build();
-  }
-}
@@ -37,6 +37,7 @@ import org.testng.annotations.Test;
 public class StreamingQueueTest extends BaseUnitTest implements Serializable {

  private static Logger LOGGER = LoggerFactory.getLogger(StreamingQueueTest.class);
+
  static {
    EnvUtil.loadNativeLibraries();
  }
@@ -56,10 +57,6 @@ public class StreamingQueueTest extends BaseUnitTest implements Serializable {
    LOGGER.warn("Do tear down");
  }

-  @BeforeClass
-  public void setUp() {
-  }
-
  @BeforeMethod
  void beforeMethod() {
    LOGGER.info("beforeTest");
@@ -79,7 +76,7 @@ public class StreamingQueueTest extends BaseUnitTest implements Serializable {
    System.clearProperty("ray.run-mode");
  }

-  @Test(timeOut = 3000000)
+  @Test(timeOut = 300000)
  public void testReaderWriter() {
    LOGGER.info("StreamingQueueTest.testReaderWriter run-mode: {}",
        System.getProperty("ray.run-mode"));
@@ -120,7 +117,7 @@ public class StreamingQueueTest extends BaseUnitTest implements Serializable {
    readerActor.call(ReaderWorker::init, inputQueueList, writerActor, msgCount);
    try {
      Thread.sleep(1000);
-    } catch (InterruptedException e) {
+    } catch (Exception e) {
      e.printStackTrace();
    }
    writerActor.call(WriterWorker::init, outputQueueList, readerActor, msgCount);
@@ -160,7 +157,7 @@ public class StreamingQueueTest extends BaseUnitTest implements Serializable {
    Map<String, Integer> wordCount = new ConcurrentHashMap<>();
    StreamingContext streamingContext = StreamingContext.buildContext();
    Map<String, String> config = new HashMap<>();
-    config.put(Config.CHANNEL_TYPE, Config.NATIVE_CHANNEL);
+    config.put(Config.CHANNEL_TYPE, "NATIVE_CHANNEL");
    config.put(Config.CHANNEL_SIZE, "100000");
    streamingContext.withConfig(config);
    List<String> text = new ArrayList<>();
@@ -4,6 +4,7 @@ import io.ray.api.BaseActorHandle;
 import io.ray.api.Ray;
 import io.ray.api.ActorHandle;
 import io.ray.runtime.functionmanager.JavaFunctionDescriptor;
+import io.ray.streaming.runtime.config.StreamingWorkerConfig;
 import io.ray.streaming.runtime.transfer.ChannelID;
 import io.ray.streaming.runtime.transfer.ChannelCreationParametersBuilder;
 import io.ray.streaming.runtime.transfer.DataMessage;
@@ -52,7 +53,7 @@ class ReaderWorker extends Worker {

  private String name = null;
  private List<String> inputQueueList = null;
-  Map<String, BaseActorHandle> fromActors = new HashMap<>();
+  Map<String, BaseActorHandle> inputActors = new HashMap<>();
  private DataReader dataReader = null;
  private long handler = 0;
  private ActorHandle<WriterWorker> peerActor = null;
@@ -87,19 +88,20 @@ class ReaderWorker extends Worker {
    LOGGER.info("java.library.path = {}", System.getProperty("java.library.path"));

    for (String queue : this.inputQueueList) {
-      fromActors.put(queue, this.peerActor);
+      inputActors.put(queue, this.peerActor);
      LOGGER.info("ReaderWorker actorId: {}", this.peerActor.getId());
    }

    Map<String, String> conf = new HashMap<>();

-    conf.put(Config.CHANNEL_TYPE, Config.NATIVE_CHANNEL);
+    conf.put(Config.CHANNEL_TYPE, "NATIVE_CHANNEL");
    conf.put(Config.CHANNEL_SIZE, "100000");
    conf.put(Config.STREAMING_JOB_NAME, "integrationTest1");
    ChannelCreationParametersBuilder.setJavaWriterFunctionDesc(
        new JavaFunctionDescriptor(Worker.class.getName(), "onWriterMessage", "([B)V"),
        new JavaFunctionDescriptor(Worker.class.getName(), "onWriterMessageSync", "([B)[B"));
-    dataReader = new DataReader(inputQueueList, fromActors, conf);
+    StreamingWorkerConfig workerConfig = new StreamingWorkerConfig(conf);
+    dataReader = new DataReader(inputQueueList, inputActors, workerConfig);

    // Should not GetBundle in RayCall thread
    Thread readThread = new Thread(Ray.wrapRunnable(new Runnable() {
@@ -171,7 +173,7 @@ class WriterWorker extends Worker {

  private String name = null;
  private List<String> outputQueueList = null;
-  Map<String, BaseActorHandle> toActors = new HashMap<>();
+  Map<String, BaseActorHandle> outputActors = new HashMap<>();
  DataWriter dataWriter = null;
  ActorHandle<ReaderWorker> peerActor = null;
  int msgCount = 0;
@@ -203,7 +205,7 @@ class WriterWorker extends Worker {
    LOGGER.info("WriterWorker init:");

    for (String queue : this.outputQueueList) {
-      toActors.put(queue, this.peerActor);
+      outputActors.put(queue, this.peerActor);
      LOGGER.info("WriterWorker actorId: {}", this.peerActor.getId());
    }

@@ -219,13 +221,14 @@ class WriterWorker extends Worker {
    }
    Map<String, String> conf = new HashMap<>();

-    conf.put(Config.CHANNEL_TYPE, Config.NATIVE_CHANNEL);
+    conf.put(Config.CHANNEL_TYPE, "NATIVE_CHANNEL");
    conf.put(Config.CHANNEL_SIZE, "100000");
    conf.put(Config.STREAMING_JOB_NAME, "integrationTest1");
    ChannelCreationParametersBuilder.setJavaReaderFunctionDesc(
        new JavaFunctionDescriptor(Worker.class.getName(), "onReaderMessage", "([B)V"),
        new JavaFunctionDescriptor(Worker.class.getName(), "onReaderMessageSync", "([B)[B"));
-    dataWriter = new DataWriter(this.outputQueueList, this.toActors, conf);
+    StreamingWorkerConfig workerConfig = new StreamingWorkerConfig(conf);
+    dataWriter = new DataWriter(outputQueueList, outputActors, workerConfig);
    Thread writerThread = new Thread(Ray.wrapRunnable(new Runnable() {
      @Override
      public void run() {
				`@@ -0,0 +1 @@`
				`io.ray.streaming.runtime.client.JobClientImpl`
				`@@ -1 +0,0 @@`
				`io.ray.streaming.runtime.schedule.JobSchedulerImpl`