/**
 * Construct job DAG, and execute the job.
 */
public void execute(String jobName) {
  JobGraphBuilder jobGraphBuilder = new JobGraphBuilder(this.streamSinks, jobName);
  JobGraph originalJobGraph = jobGraphBuilder.build();
  this.jobGraph = new JobGraphOptimizer(originalJobGraph).optimize();
  jobGraph.printJobGraph();
  LOG.info("JobGraph digraph\n{}", jobGraph.generateDigraph());

  if (!Ray.isInitialized()) {
    if (Config.MEMORY_CHANNEL.equalsIgnoreCase(jobConfig.get(Config.CHANNEL_TYPE))) {
      Preconditions.checkArgument(!jobGraph.isCrossLanguageGraph());
      ClusterStarter.startCluster(false, true);
      LOG.info("Created local cluster for job {}.", jobName);
    } else {
      ClusterStarter.startCluster(jobGraph.isCrossLanguageGraph(), false);
      LOG.info("Created multi process cluster for job {}.", jobName);
    }
    Runtime.getRuntime().addShutdownHook(new Thread(StreamingContext.this::stop));
  } else {
    LOG.info("Reuse existing cluster.");
  }

  ServiceLoader<JobClient> serviceLoader = ServiceLoader.load(JobClient.class);
  Iterator<JobClient> iterator = serviceLoader.iterator();
  Preconditions.checkArgument(iterator.hasNext(),
      "No JobClient implementation has been provided.");
  JobClient jobClient = iterator.next();
  jobClient.submit(jobGraph, jobConfig);
}

public boolean submitJob(ActorHandle<JobMaster> jobMasterActor, JobGraph jobGraph) {
  LOG.info("Begin submitting job using logical plan: {}.", jobGraph);

  this.jobMasterActor = jobMasterActor;

  // init manager
  graphManager = new GraphManagerImpl(runtimeContext);
  resourceManager = new ResourceManagerImpl(runtimeContext);

  // build and set graph into runtime context
  ExecutionGraph executionGraph = graphManager.buildExecutionGraph(jobGraph);
  runtimeContext.setJobGraph(jobGraph);
  runtimeContext.setExecutionGraph(executionGraph);

  // init scheduler
  try {
    scheduler = new JobSchedulerImpl(this);
    scheduler.scheduleJob(graphManager.getExecutionGraph());
  } catch (Exception e) {
    LOG.error("Failed to submit job.", e);
    return false;
  }
  return true;
}

// init coordinators
checkpointCoordinator = new CheckpointCoordinator(this);
checkpointCoordinator.start();
failoverCoordinator = new FailoverCoordinator(this, isRecover);
failoverCoordinator.start();

saveContext();

public byte[] reportJobWorkerCommit(byte[] reportBytes) {
  Boolean ret = false;
  RemoteCall.BaseWorkerCmd reportPb;
  try {
    reportPb = RemoteCall.BaseWorkerCmd.parseFrom(reportBytes);
    ActorId actorId = ActorId.fromBytes(reportPb.getActorId().toByteArray());
    long remoteCallCost = System.currentTimeMillis() - reportPb.getTimestamp();
    LOG.info("Vertex {}, request job worker commit cost {}ms, actorId={}.",
        getExecutionVertex(actorId), remoteCallCost, actorId);
    RemoteCall.WorkerCommitReport commit =
        reportPb.getDetail().unpack(RemoteCall.WorkerCommitReport.class);
    WorkerCommitReport report = new WorkerCommitReport(actorId, commit.getCommitCheckpointId());
    ret = checkpointCoordinator.reportJobWorkerCommit(report);
  } catch (InvalidProtocolBufferException e) {
    LOG.error("Parse job worker commit has exception.", e);
  }
  return RemoteCall.BoolResult.newBuilder().setBoolRes(ret).build().toByteArray();
}

public byte[] requestJobWorkerRollback(byte[] requestBytes) {
    Boolean ret = false;
    RemoteCall.BaseWorkerCmd requestPb;
    try {
      requestPb = RemoteCall.BaseWorkerCmd.parseFrom(requestBytes);
      ActorId actorId = ActorId.fromBytes(requestPb.getActorId().toByteArray());
      long remoteCallCost = System.currentTimeMillis() - requestPb.getTimestamp();
      ExecutionGraph executionGraph = graphManager.getExecutionGraph();
      Optional<BaseActorHandle> rayActor = executionGraph.getActorById(actorId);
      if (!rayActor.isPresent()) {
        LOG.warn("Skip this invalid rollback, actor id {} is not found.", actorId);
        return RemoteCall.BoolResult.newBuilder().setBoolRes(false).build().toByteArray();
      }
      ExecutionVertex exeVertex = getExecutionVertex(actorId);
      LOG.info("Vertex {}, request job worker rollback cost {}ms, actorId={}.",
          exeVertex, remoteCallCost, actorId);
      RemoteCall.WorkerRollbackRequest rollbackPb
          = RemoteCall.WorkerRollbackRequest.parseFrom(requestPb.getDetail().getValue());
      exeVertex.setPid(rollbackPb.getWorkerPid());
      // To find old container where slot is located in.
      String hostname = "";
      Optional<Container> container = ResourceUtil.getContainerById(
          resourceManager.getRegisteredContainers(),
          exeVertex.getContainerId()
      );
      if (container.isPresent()) {
        hostname = container.get().getHostname();
      }
      WorkerRollbackRequest request = new WorkerRollbackRequest(
          actorId, rollbackPb.getExceptionMsg(), hostname, exeVertex.getPid()
      );

      ret = failoverCoordinator.requestJobWorkerRollback(request);
      LOG.info("Vertex {} request rollback, exception msg : {}.",
          exeVertex, rollbackPb.getExceptionMsg());

    } catch (Throwable e) {
      LOG.error("Parse job worker rollback has exception.", e);
    }
    return RemoteCall.BoolResult.newBuilder().setBoolRes(ret).build().toByteArray();
  }

private boolean createWorker(ExecutionVertex executionVertex) {
  LOG.info("Start to create worker actor for vertex: {} with resource: {}, workeConfig: {}.",
      executionVertex.getExecutionVertexName(), executionVertex.getResource(),
      executionVertex.getWorkerConfig());

  Language language = executionVertex.getLanguage();

  BaseActorHandle actor;
  if (Language.JAVA == language) {
    actor = Ray.actor(JobWorker::new, executionVertex)
        .setResources(executionVertex.getResource())
        .setMaxRestarts(-1)
        .remote();
  } else {
    RemoteCall.ExecutionVertexContext.ExecutionVertex vertexPb
        = new GraphPbBuilder().buildVertex(executionVertex);
    actor = Ray.actor(
        PyActorClass.of("ray.streaming.runtime.worker", "JobWorker"), vertexPb.toByteArray())
        .setResources(executionVertex.getResource())
        .setMaxRestarts(-1)
        .remote();
  }

  if (null == actor) {
    LOG.error("Create worker actor failed.");
    return false;
  }

  executionVertex.setWorkerActor(actor);

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

public boolean initWorkers(
    Map<ExecutionVertex, JobWorkerContext> vertexToContextMap, int timeout) {
  LOG.info("Begin initiating workers: {}.", vertexToContextMap);
  long startTime = System.currentTimeMillis();

  Map<ObjectRef<Boolean>, ActorId> rayObjects = new HashMap<>();
  vertexToContextMap.entrySet().forEach((entry -> {
    ExecutionVertex vertex = entry.getKey();
    rayObjects.put(RemoteCallWorker.initWorker(vertex.getWorkerActor(), entry.getValue()),
        vertex.getWorkerActorId());
  }));

  List<ObjectRef<Boolean>> objectRefList = new ArrayList<>(rayObjects.keySet());

  LOG.info("Waiting for workers' initialization.");
  WaitResult<Boolean> result = Ray.wait(objectRefList, objectRefList.size(), timeout);
  if (result.getReady().size() != objectRefList.size()) {
    LOG.error("Initializing workers timeout[{} ms].", timeout);
    return false;
  }

  LOG.info("Finished waiting workers' initialization.");
  LOG.info("Workers initialized. Cost {} ms.", System.currentTimeMillis() - startTime);
  return true;
}

public boolean startWorkers(ExecutionGraph executionGraph, long lastCheckpointId, int timeout) {
    LOG.info("Begin starting workers.");
    long startTime = System.currentTimeMillis();
    List<ObjectRef<Object>> objectRefs = new ArrayList<>();

    // start source actors 1st
    executionGraph.getSourceActors()
        .forEach(actor -> objectRefs.add(RemoteCallWorker.rollback(actor, lastCheckpointId)));

    // then start non-source actors
    executionGraph.getNonSourceActors()
        .forEach(actor -> objectRefs.add(RemoteCallWorker.rollback(actor, lastCheckpointId)));

    WaitResult<Object> result = Ray.wait(objectRefs, objectRefs.size(), timeout);
    if (result.getReady().size() != objectRefs.size()) {
      LOG.error("Starting workers timeout[{} ms].", timeout);
      return false;
    }

    LOG.info("Workers started. Cost {} ms.", System.currentTimeMillis() - startTime);
    return true;
  }

public static ObjectRef<Boolean> initWorker(BaseActorHandle actor, JobWorkerContext context) {
  LOG.info("Call worker to initiate, actor: {}, context: {}.", actor.getId(), context);
  ObjectRef<Boolean> result;

  // python
  if (actor instanceof PyActorHandle) {
    result = ((PyActorHandle) actor).task(PyActorMethod.of("init", Boolean.class),
        context.getPythonWorkerContextBytes()).remote();
  } else {
    // java
    result = ((ActorHandle<JobWorker>) actor).task(JobWorker::init, context).remote();
  }

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

public static ObjectRef rollback(BaseActorHandle actor, final Long checkpointId) {
  LOG.info("Call worker to start, actor: {}.", actor.getId());
  ObjectRef result;

  // python
  if (actor instanceof PyActorHandle) {
    RemoteCall.CheckpointId checkpointIdPb = RemoteCall.CheckpointId.newBuilder()
        .setCheckpointId(checkpointId)
        .build();
    result = ((PyActorHandle) actor)
        .task(PyActorMethod.of("rollback"),
            checkpointIdPb.toByteArray()
        ).remote();
  } else {
    // java
    result = ((ActorHandle<JobWorker>) actor)
        .task(JobWorker::rollback, checkpointId, System.currentTimeMillis()).remote();
  }

  LOG.info("Finished calling worker to start.");
  return result;
}

public static ObjectRef<byte[]> reportJobWorkerCommitAsync(
    ActorHandle<JobMaster> actor,
    WorkerCommitReport commitReport) {
  RemoteCall.WorkerCommitReport commit = RemoteCall.WorkerCommitReport.newBuilder()
      .setCommitCheckpointId(commitReport.commitCheckpointId)
      .build();
  Any detail = Any.pack(commit);
  RemoteCall.BaseWorkerCmd cmd = RemoteCall.BaseWorkerCmd.newBuilder()
      .setActorId(ByteString.copyFrom(commitReport.fromActorId.getBytes()))
      .setTimestamp(System.currentTimeMillis())
      .setDetail(detail).build();

  return actor.task(JobMaster::reportJobWorkerCommit, cmd.toByteArray()).remote();
}

public static Boolean requestJobWorkerRollback(
      ActorHandle<JobMaster> actor,
      WorkerRollbackRequest rollbackRequest) {
    RemoteCall.WorkerRollbackRequest request = RemoteCall.WorkerRollbackRequest.newBuilder()
        .setExceptionMsg(rollbackRequest.getRollbackExceptionMsg())
        .setWorkerHostname(rollbackRequest.getHostname())
        .setWorkerPid(rollbackRequest.getPid()).build();
    Any detail = Any.pack(request);
    RemoteCall.BaseWorkerCmd cmd = RemoteCall.BaseWorkerCmd.newBuilder()
        .setActorId(ByteString.copyFrom(rollbackRequest.fromActorId.getBytes()))
        .setTimestamp(System.currentTimeMillis())
        .setDetail(detail).build();
    ObjectRef<byte[]> ret = actor.task(
        JobMaster::requestJobWorkerRollback, cmd.toByteArray()).remote();
    byte[] res = ret.get();
    return PbResultParser.parseBoolResult(res);
  }

public JobWorker(ExecutionVertex executionVertex) {
  LOG.info("Creating job worker.");

  // TODO: the following 3 lines is duplicated with that in init(), try to optimise it later.
  this.executionVertex = executionVertex;
  this.workerConfig = new StreamingWorkerConfig(executionVertex.getWorkerConfig());
  this.contextBackend = ContextBackendFactory.getContextBackend(this.workerConfig);

  LOG.info("Ray.getRuntimeContext().wasCurrentActorRestarted()={}",
      Ray.getRuntimeContext().wasCurrentActorRestarted());
  if (!Ray.getRuntimeContext().wasCurrentActorRestarted()) {
    saveContext();
    LOG.info("Job worker is fresh started, init success.");
    return;
  }

  LOG.info("Begin load job worker checkpoint state.");

  byte[] bytes = CheckpointStateUtil.get(contextBackend, getJobWorkerContextKey());
  if (bytes != null) {
    JobWorkerContext context = Serializer.decode(bytes);
    LOG.info("Worker recover from checkpoint state, byte len={}, context={}.", bytes.length,
        context);
    init(context);
    requestRollback("LoadCheckpoint request rollback in new actor.");
  } else {
    LOG.error(
        "Worker is reconstructed, but can't load checkpoint. " +
            "Check whether you checkpoint state is reliable. Current checkpoint state is {}.",
        contextBackend.getClass().getName());
  }
}

public Boolean init(JobWorkerContext workerContext) {
  // IMPORTANT: some test cases depends on this log to find workers' pid,
  // be careful when changing this log.
  LOG.info("Initiating job worker: {}. Worker context is: {}, pid={}.",
      workerContext.getWorkerName(), workerContext, EnvUtil.getJvmPid());

  this.workerContext = workerContext;
  this.executionVertex = workerContext.getExecutionVertex();
  this.workerConfig = new StreamingWorkerConfig(executionVertex.getWorkerConfig());
  // init state backend
  this.contextBackend = ContextBackendFactory.getContextBackend(this.workerConfig);

  LOG.info("Initiating job worker succeeded: {}.", workerContext.getWorkerName());
  saveContext();
  return true;
}

public synchronized void saveContext() {
  byte[] contextBytes = Serializer.encode(workerContext);
  String key = getJobWorkerContextKey();
  LOG.info("Saving context, worker context={}, serialized byte length={}, key={}.", workerContext,
      contextBytes.length, key);
  CheckpointStateUtil.put(contextBackend, key, contextBytes);
}

public CallResult<ChannelRecoverInfo> rollback(Long checkpointId, Long startRollbackTs) {
   synchronized (initialStateChangeLock) {
     if (task != null && task.isAlive() && checkpointId == task.lastCheckpointId &&
         task.isInitialState) {
       return CallResult.skipped("Task is already in initial state, skip this rollback.");
     }
   }
   long remoteCallCost = System.currentTimeMillis() - startRollbackTs;

   LOG.info("Start rollback[{}], checkpoint is {}, remote call cost {}ms.",
       executionVertex.getExecutionJobVertexName(), checkpointId, remoteCallCost);

   rollbackCount++;
   if (rollbackCount > 1) {
     isRecreate.set(true);
   }

   try {
     //Init transfer
     TransferChannelType channelType = workerConfig.transferConfig.channelType();
     if (TransferChannelType.NATIVE_CHANNEL == channelType) {
       transferHandler = new TransferHandler();
     }

     if (task != null) {
       // make sure the task is closed
       task.close();
       task = null;
     }

     // create stream task
     task = createStreamTask(checkpointId);
     ChannelRecoverInfo channelRecoverInfo = task.recover(isRecreate.get());
     isNeedRollback = false;

     LOG.info("Rollback job worker success, checkpoint is {}, channelRecoverInfo is {}.",
         checkpointId, channelRecoverInfo);

     return CallResult.success(channelRecoverInfo);
   } catch (Exception e) {
     LOG.error("Rollback job worker has exception.", e);
     return CallResult.fail(ExceptionUtils.getStackTrace(e));
   }
 }

 private StreamTask createStreamTask(long checkpointId) {
   StreamTask task;
   StreamProcessor streamProcessor = ProcessBuilder
       .buildProcessor(executionVertex.getStreamOperator());
   LOG.debug("Stream processor created: {}.", streamProcessor);

   if (streamProcessor instanceof SourceProcessor) {
     task = new SourceStreamTask(streamProcessor, this, checkpointId);
   } else if (streamProcessor instanceof OneInputProcessor) {
     task = new OneInputStreamTask(streamProcessor, this, checkpointId);
   } else {
     throw new RuntimeException("Unsupported processor type:" + streamProcessor);
   }
   LOG.info("Stream task created: {}.", task);
   return task;
 }

public Boolean triggerCheckpoint(Long barrierId) {
  LOG.info("Receive trigger, barrierId is {}.", barrierId);
  if (task != null) {
    return task.triggerCheckpoint(barrierId);
  }
  return false;
}

public Boolean notifyCheckpointTimeout(Long checkpointId) {
  LOG.info("Notify checkpoint timeout, checkpoint id is {}.", checkpointId);
  if (task != null) {
    task.notifyCheckpointTimeout(checkpointId);
  }
  return true;
}

public Boolean clearExpiredCheckpoint(Long expiredStateCpId, Long expiredQueueCpId) {
  LOG.info("Clear expired checkpoint state, checkpoint id is {}; " +
          "Clear expired queue msg, checkpoint id is {}",
      expiredStateCpId, expiredQueueCpId);
  if (task != null) {
    if (expiredStateCpId > 0) {
      task.clearExpiredCpState(expiredStateCpId);
    }
    task.clearExpiredQueueMsg(expiredQueueCpId);
  }
  return true;
}

public void requestRollback(String exceptionMsg) {
  LOG.info("Request rollback.");
  isNeedRollback = true;
  isRecreate.set(true);
  boolean requestRet = RemoteCallMaster.requestJobWorkerRollback(
      workerContext.getMaster(), new WorkerRollbackRequest(
          workerContext.getWorkerActorId(),
          exceptionMsg,
          EnvUtil.getHostName(),
          EnvUtil.getJvmPid()
      ));
  if (!requestRet) {
    LOG.warn("Job worker request rollback failed! exceptionMsg={}.", exceptionMsg);
  }
}

public Boolean checkIfNeedRollback(Long startCallTs) {
  // No save checkpoint in this query.
  long remoteCallCost = System.currentTimeMillis() - startCallTs;
  LOG.info("Finished checking if need to rollback with result: {}, rpc delay={}ms.",
      isNeedRollback, remoteCallCost);
  return isNeedRollback;
}

protected StreamTask(Processor processor, JobWorker jobWorker, long lastCheckpointId) {
  this.processor = processor;
  this.jobWorker = jobWorker;
  this.checkpointState = jobWorker.contextBackend;
  this.lastCheckpointId = lastCheckpointId;

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

public ChannelRecoverInfo recover(boolean isRecover) {

  if (isRecover) {
    LOG.info("Stream task begin recover.");
  } else {
    LOG.info("Stream task first start begin.");
  }
  prepareTask(isRecover);

  // start runner
  ChannelRecoverInfo recoverInfo = new ChannelRecoverInfo(new HashMap<>());
  if (reader != null) {
    recoverInfo = reader.getQueueRecoverInfo();
  }

  thread.setUncaughtExceptionHandler((t, e) -> LOG.error("Uncaught exception in runner thread.", e));
  LOG.info("Start stream task: {}.", this.getClass().getSimpleName());
  thread.start();

  if (isRecover) {
    LOG.info("Stream task recover end.");
  } else {
    LOG.info("Stream task first start finished.");
  }

  return recoverInfo;
}

private void prepareTask(boolean isRecreate) {
  LOG.info("Preparing stream task, isRecreate={}.", isRecreate);
  ExecutionVertex executionVertex = jobWorker.getExecutionVertex();

  // 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());

  OperatorCheckpointInfo operatorCheckpointInfo = new OperatorCheckpointInfo();
  byte[] bytes = null;

  // Fetch checkpoint from storage only in recreate mode not for new startup
  // worker
  // in rescaling or something like that.
  if (isRecreate) {
    String cpKey = genOpCheckpointKey(lastCheckpointId);
    LOG.info("Getting task checkpoints from state, cpKey={}, checkpointId={}.", cpKey, lastCheckpointId);
    bytes = CheckpointStateUtil.get(checkpointState, cpKey);
    if (bytes == null) {
      String msg = String.format("Task recover failed, checkpoint is null! cpKey=%s", cpKey);
      throw new RuntimeException(msg);
    }
  }

  // when use memory state, if actor throw exception, will miss state
  if (bytes != null) {
    operatorCheckpointInfo = Serializer.decode(bytes);
    processor.loadCheckpoint(operatorCheckpointInfo.processorCheckpoint);
    LOG.info("Stream task recover from checkpoint state, checkpoint bytes len={}, checkpointInfo={}.", bytes.length,
        operatorCheckpointInfo);
  }

  // writer
  if (!executionVertex.getOutputEdges().isEmpty()) {
    LOG.info("Register queue writer, channels={}, outputCheckpoints={}.", executionVertex.getOutputChannelIdList(),
        operatorCheckpointInfo.outputPoints);
    writer = new DataWriter(executionVertex.getOutputChannelIdList(), executionVertex.getOutputActorList(),
        operatorCheckpointInfo.outputPoints, jobWorker.getWorkerConfig());
  }

  // reader
  if (!executionVertex.getInputEdges().isEmpty()) {
    LOG.info("Register queue reader, channels={}, inputCheckpoints={}.", executionVertex.getInputChannelIdList(),
        operatorCheckpointInfo.inputPoints);
    reader = new DataReader(executionVertex.getInputChannelIdList(), executionVertex.getInputActorList(),
        operatorCheckpointInfo.inputPoints, jobWorker.getWorkerConfig());
  }

  openProcessor();

  LOG.debug("Finished preparing stream task.");
}

private void openProcessor() {
  ExecutionVertex executionVertex = jobWorker.getExecutionVertex();
  List<ExecutionEdge> outputEdges = executionVertex.getOutputEdges();

  Map<String, List<String>> opGroupedChannelId = new HashMap<>();
  Map<String, List<BaseActorHandle>> opGroupedActor = new HashMap<>();
  Map<String, Partition> opPartitionMap = new HashMap<>();
  for (int i = 0; i < outputEdges.size(); ++i) {
    ExecutionEdge edge = outputEdges.get(i);
    String opName = edge.getTargetExecutionJobVertexName();
    if (!opPartitionMap.containsKey(opName)) {
      opGroupedChannelId.put(opName, new ArrayList<>());
      opGroupedActor.put(opName, new ArrayList<>());
    }
    opGroupedChannelId.get(opName).add(executionVertex.getOutputChannelIdList().get(i));
    opGroupedActor.get(opName).add(executionVertex.getOutputActorList().get(i));
    opPartitionMap.put(opName, edge.getPartition());
  }
  opPartitionMap.keySet().forEach(opName -> {
    collectors.add(new OutputCollector(writer, opGroupedChannelId.get(opName), opGroupedActor.get(opName),
        opPartitionMap.get(opName)));
  });

  RuntimeContext runtimeContext = new StreamingRuntimeContext(executionVertex, jobWorker.getWorkerConfig().configMap,
      executionVertex.getParallelism());

  processor.open(collectors, runtimeContext);
}

public boolean triggerCheckpoint(Long barrierId) {
  throw new UnsupportedOperationException("Only source operator supports trigger checkpoints.");
}

public void doCheckpoint(long checkpointId, Map<String, OffsetInfo> inputPoints) {
  Map<String, OffsetInfo> outputPoints = null;
  if (writer != null) {
    outputPoints = writer.getOutputCheckpoints();
    RemoteCall.Barrier barrierPb = RemoteCall.Barrier.newBuilder().setId(checkpointId).build();
    ByteBuffer byteBuffer = ByteBuffer.wrap(barrierPb.toByteArray());
    byteBuffer.order(ByteOrder.nativeOrder());
    writer.broadcastBarrier(checkpointId, byteBuffer);
  }

  LOG.info("Start do checkpoint, cp id={}, inputPoints={}, outputPoints={}.", checkpointId, inputPoints,
      outputPoints);

  this.lastCheckpointId = checkpointId;
  Serializable processorCheckpoint = processor.saveCheckpoint();

  try {
    OperatorCheckpointInfo opCpInfo = new OperatorCheckpointInfo(inputPoints, outputPoints, processorCheckpoint,
        checkpointId);
    saveCpStateAndReport(opCpInfo, checkpointId);
  } catch (Exception e) {
    // there will be exceptions when flush state to backend.
    // we ignore the exception to prevent failover
    LOG.error("Processor or op checkpoint exception.", e);
  }

  LOG.info("Operator do checkpoint {} finish.", checkpointId);
}

private void saveCpStateAndReport(OperatorCheckpointInfo operatorCheckpointInfo, long checkpointId) {
  saveCp(operatorCheckpointInfo, checkpointId);
  reportCommit(checkpointId);

  LOG.info("Finish save cp state and report, checkpoint id is {}.", checkpointId);
}

private void saveCp(OperatorCheckpointInfo operatorCheckpointInfo, long checkpointId) {
  byte[] bytes = Serializer.encode(operatorCheckpointInfo);
  String cpKey = genOpCheckpointKey(checkpointId);
  LOG.info("Saving task checkpoint, cpKey={}, byte len={}, checkpointInfo={}.", cpKey, bytes.length,
      operatorCheckpointInfo);
  synchronized (checkpointState) {
    if (outdatedCheckpoints.contains(checkpointId)) {
      LOG.info("Outdated checkpoint, skip save checkpoint.");
      outdatedCheckpoints.remove(checkpointId);
    } else {
      CheckpointStateUtil.put(checkpointState, cpKey, bytes);
    }
  }
}

private void reportCommit(long checkpointId) {
  final JobWorkerContext context = jobWorker.getWorkerContext();
  LOG.info("Report commit async, checkpoint id {}.", checkpointId);
  RemoteCallMaster.reportJobWorkerCommitAsync(context.getMaster(),
      new WorkerCommitReport(context.getWorkerActorId(), checkpointId));
}

public void notifyCheckpointTimeout(long checkpointId) {
  String cpKey = genOpCheckpointKey(checkpointId);
  try {
    synchronized (checkpointState) {
      if (checkpointState.exists(cpKey)) {
        checkpointState.remove(cpKey);
      } else {
        outdatedCheckpoints.add(checkpointId);
      }
    }
  } catch (Exception e) {
    LOG.error("Notify checkpoint timeout failed, checkpointId is {}.", checkpointId, e);
  }
}

public void clearExpiredCpState(long checkpointId) {
  String cpKey = genOpCheckpointKey(checkpointId);
  try {
    checkpointState.remove(cpKey);
  } catch (Exception e) {
    LOG.error("Failed to remove key {} from state backend.", cpKey, e);
  }
}

public void clearExpiredQueueMsg(long checkpointId) {
  // get operator checkpoint
  String cpKey = genOpCheckpointKey(checkpointId);
  byte[] bytes;
  try {
    bytes = checkpointState.get(cpKey);
  } catch (Exception e) {
    LOG.error("Failed to get key {} from state backend.", cpKey, e);
    return;
  }
  if (bytes != null) {
    final OperatorCheckpointInfo operatorCheckpointInfo = Serializer.decode(bytes);
    long cpId = operatorCheckpointInfo.checkpointId;
    if (writer != null) {
      writer.clearCheckpoint(cpId);
    }
  }
}

protected void requestRollback(String exceptionMsg) {
  jobWorker.requestRollback(exceptionMsg);
}

public void run() {
  LOG.info("Source stream task thread start.");
  Long barrierId;
  try {
    while (running) {
      isInitialState = false;

      // check checkpoint
      barrierId = pendingBarrier.get();
      if (barrierId != null) {
        // Important: because cp maybe timeout, master will use the old checkpoint id again
        if (pendingBarrier.compareAndSet(barrierId, null)) {
          // source fetcher only have outputPoints
          LOG.info("Start to do checkpoint {}, worker name is {}.",
              barrierId, jobWorker.getWorkerContext().getWorkerName());

          doCheckpoint(barrierId, null);

          LOG.info("Finish to do checkpoint {}.", barrierId);
        } else {
          // pendingCheckpointId has modify, should not happen
          LOG.warn("Pending checkpointId modify unexpected, expect={}, now={}.", barrierId,
              pendingBarrier.get());
        }
      }

      sourceProcessor.fetch();
    }
  } catch (Throwable e) {
    if (e instanceof ChannelInterruptException ||
        ExceptionUtils.getRootCause(e) instanceof ChannelInterruptException) {
      LOG.info("queue has stopped.");
    } else {
      // occur error, need to rollback
      LOG.error("Last success checkpointId={}, now occur error.", lastCheckpointId, e);
      requestRollback(ExceptionUtils.getStackTrace(e));
    }
  }

  LOG.info("Source stream task thread exit.");
}

public boolean triggerCheckpoint(Long barrierId) {
  return pendingBarrier.compareAndSet(null, barrierId);
}

public void run() {
   try {
     while (running) {
       ChannelMessage item;

       // reader.read() will change the consumer state once it got an item. This lock is to
       // ensure worker can get correct isInitialState value in exactly-once-mode's rollback.
       synchronized (jobWorker.initialStateChangeLock) {
         item = reader.read(readTimeoutMillis);
         if (item != null) {
           isInitialState = false;
         } else {
           continue;
         }
       }

       if (item instanceof DataMessage) {
         DataMessage dataMessage = (DataMessage) item;
         byte[] bytes = new byte[dataMessage.body().remaining() - 1];
         byte typeId = dataMessage.body().get();
         dataMessage.body().get(bytes);
         Object obj;
         if (typeId == Serializer.JAVA_TYPE_ID) {
           obj = javaSerializer.deserialize(bytes);
         } else {
           obj = crossLangSerializer.deserialize(bytes);
         }
         processor.process(obj);
       } else if (item instanceof BarrierMessage) {
         final BarrierMessage queueBarrier = (BarrierMessage) item;
         byte[] barrierData = new byte[queueBarrier.getData().remaining()];
         queueBarrier.getData().get(barrierData);
         RemoteCall.Barrier barrierPb = RemoteCall.Barrier.parseFrom(barrierData);
         final long checkpointId = barrierPb.getId();
         LOG.info("Start to do checkpoint {}, worker name is {}.", checkpointId,
             jobWorker.getWorkerContext().getWorkerName());

         final Map<String, OffsetInfo> inputPoints = queueBarrier.getInputOffsets();
         doCheckpoint(checkpointId, inputPoints);
         LOG.info("Do checkpoint {} success.", checkpointId);
       }
     }
   } catch (Throwable throwable) {
     if (throwable instanceof ChannelInterruptException ||
         ExceptionUtils.getRootCause(throwable) instanceof ChannelInterruptException) {
       LOG.info("queue has stopped.");
     } else {
       // error occurred, need to rollback
       LOG.error("Last success checkpointId={}, now occur error.", lastCheckpointId, throwable);
       requestRollback(ExceptionUtils.getStackTrace(throwable));
     }
   }
   LOG.info("Input stream task thread exit.");
   stopped = true;
 }

public StreamingRuntimeContext(
    ExecutionVertex executionVertex, Map<String, String> config,
    int parallelism) {
  this.taskId = executionVertex.getExecutionVertexId();
  this.config = config;
  this.taskIndex = executionVertex.getExecutionVertexIndex();
  this.parallelism = parallelism;
}

void open(List<Collector> collectors, RuntimeContext runtimeContext);

void process(T t);

/**
 * See {@link Function#saveCheckpoint()}.
 */
Serializable saveCheckpoint();

/**
 * See {@link Function#loadCheckpoint(Serializable)}.
 */
void loadCheckpoint(Serializable checkpointObject);

void close();

/**
 * check if key exists in state
 *
 * @return true if exists
 */
boolean exists(final String key) throws Exception;

/**
 * get content by key
 *
 * @param key key
 * @return the StateBackend
 */
byte[] get(final String key) throws Exception;

/**
 * put content by key
 *
 * @param key key
 * @param value content
 */
void put(final String key, final byte[] value) throws Exception;

/**
 * remove content by key
 *
 * @param key key
 */
void remove(final String key) throws Exception;

public BaseCoordinator(JobMaster jobMaster) {
  this.jobMaster = jobMaster;
  this.runtimeContext = jobMaster.getRuntimeContext();
  this.graphManager = jobMaster.getGraphManager();
}

public void start() {
  thread = new Thread(Ray.wrapRunnable(this),
      this.getClass().getName() + "-" + System.currentTimeMillis());
  thread.start();
}

public void stop() {
  closed = true;

  try {
    if (thread != null) {
      thread.join(30000);
    }
  } catch (InterruptedException e) {
    LOG.error("Coordinator thread exit has exception.", e);
  }
}

public void run() {
  while (!closed) {
    try {
      final BaseWorkerCmd command = runtimeContext.cpCmds.poll(1, TimeUnit.SECONDS);
      if (command != null) {
        if (command instanceof WorkerCommitReport) {
          processCommitReport((WorkerCommitReport) command);
        } else {
          interruptCheckpoint();
        }
      }

      if (!pendingCheckpointActors.isEmpty()) {
        // if wait commit report timeout, this cp fail, and restart next cp
        if (timeoutOnWaitCheckpoint()) {
          LOG.warn("Waiting for checkpoint {} timeout, pending cp actors is {}.",
              runtimeContext.lastCheckpointId,
              graphManager.getExecutionGraph().getActorName(pendingCheckpointActors));

          interruptCheckpoint();
        }
      } else {
        maybeTriggerCheckpoint();
      }
    } catch (Throwable e) {
      LOG.error("Checkpoint coordinator occur err.", e);
      try {
        interruptCheckpoint();
      } catch (Throwable interruptE) {
        LOG.error("Ignore interrupt checkpoint exception in catch block.");
      }
    }
  }
  LOG.warn("Checkpoint coordinator thread exit.");
}

private void processCommitReport(WorkerCommitReport commitReport) {
  LOG.info("Start process commit report {}, from actor name={}.", commitReport,
      graphManager.getExecutionGraph().getActorName(commitReport.fromActorId));

  try {
    Preconditions.checkArgument(
        commitReport.commitCheckpointId == runtimeContext.lastCheckpointId,
        "expect checkpointId %s, but got %s",
        runtimeContext.lastCheckpointId, commitReport);

    if (!pendingCheckpointActors.contains(commitReport.fromActorId)) {
      LOG.warn("Invalid commit report, skipped.");
      return;
    }

    pendingCheckpointActors.remove(commitReport.fromActorId);
    LOG.info("Pending actors after this commit: {}.",
        graphManager.getExecutionGraph().getActorName(pendingCheckpointActors));

    // checkpoint finish
    if (pendingCheckpointActors.isEmpty()) {
      // actor finish
      runtimeContext.checkpointIds.add(runtimeContext.lastCheckpointId);

      if (clearExpiredCpStateAndQueueMsg()) {
        // save master context
        jobMaster.saveContext();

        LOG.info("Finish checkpoint: {}.", runtimeContext.lastCheckpointId);
      } else {
        LOG.warn("Fail to do checkpoint: {}.", runtimeContext.lastCheckpointId);
      }
    }

    LOG.info("Process commit report {} success.", commitReport);
  } catch (Throwable e) {
    LOG.warn("Process commit report has exception.", e);
  }
}

private boolean clearExpiredCpStateAndQueueMsg() {
  // queue msg must clear when first checkpoint finish
  List<BaseActorHandle> allActor = graphManager.getExecutionGraph().getAllActors();
  if (1 == runtimeContext.checkpointIds.size()) {
    Long msgExpiredCheckpointId = runtimeContext.checkpointIds.get(0);
    RemoteCallWorker.clearExpiredCheckpointParallel(allActor, 0L, msgExpiredCheckpointId);
  }

  if (runtimeContext.checkpointIds.size() > 1) {
    Long stateExpiredCpId = runtimeContext.checkpointIds.remove(0);
    Long msgExpiredCheckpointId = runtimeContext.checkpointIds.get(0);
    RemoteCallWorker
        .clearExpiredCheckpointParallel(allActor, stateExpiredCpId, msgExpiredCheckpointId);
  }
  return true;
}

private void interruptCheckpoint() {
  // notify checkpoint timeout is time-consuming while many workers crash or
  // container failover.
  if (interruptedCheckpointSet.contains(runtimeContext.lastCheckpointId)) {
    LOG.warn("Skip interrupt duplicated checkpoint id : {}.", runtimeContext.lastCheckpointId);
    return;
  }
  interruptedCheckpointSet.add(runtimeContext.lastCheckpointId);
  LOG.warn("Interrupt checkpoint, checkpoint id : {}.", runtimeContext.lastCheckpointId);

  List<BaseActorHandle> allActor = graphManager.getExecutionGraph().getAllActors();
  if (runtimeContext.lastCheckpointId > runtimeContext.getLastValidCheckpointId()) {
    RemoteCallWorker
        .notifyCheckpointTimeoutParallel(allActor, runtimeContext.lastCheckpointId);
  }

  if (!pendingCheckpointActors.isEmpty()) {
    pendingCheckpointActors.clear();
  }
  maybeTriggerCheckpoint();
}

private void maybeTriggerCheckpoint() {
  if (readyToTrigger()) {
    triggerCheckpoint();
  }
}

private boolean readyToTrigger() {
  return (System.currentTimeMillis() - runtimeContext.lastCpTimestamp) >=
      cpIntervalSecs * 1000;
}

private void triggerCheckpoint() {
  interruptedCheckpointSet.clear();
  if (LOG.isInfoEnabled()) {
    LOG.info("Start trigger checkpoint {}.", runtimeContext.lastCheckpointId + 1);
  }

  List<ActorId> allIds = graphManager.getExecutionGraph().getAllActorsId();
  // do the checkpoint
  pendingCheckpointActors.addAll(allIds);

  // inc last checkpoint id
  ++runtimeContext.lastCheckpointId;

  final List<ObjectRef> sourcesRet = new ArrayList<>();

  graphManager.getExecutionGraph().getSourceActors().forEach(actor -> {
    sourcesRet.add(RemoteCallWorker.triggerCheckpoint(
        actor, runtimeContext.lastCheckpointId));
  });

  for (ObjectRef rayObject : sourcesRet) {
    if (rayObject.get() instanceof RayException) {
      LOG.warn("Trigger checkpoint has exception.", (RayException) rayObject.get());
      throw (RayException) rayObject.get();
    }
  }
  runtimeContext.lastCpTimestamp = System.currentTimeMillis();
  LOG.info("Trigger checkpoint success.");
}

public Boolean requestJobWorkerRollback(WorkerRollbackRequest request) {
  LOG.info("Request job worker rollback {}.", request);
  boolean ret;
  if (!isDuplicateRequest(request)) {
    ret = runtimeContext.foCmds.offer(request);
  } else {
    LOG.warn("Skip duplicated worker rollback request, {}.", request.toString());
    return true;
  }
  jobMaster.saveContext();
  if (!ret) {
    LOG.warn("Request job worker rollback failed, because command queue is full.");
  }
  return ret;
}

public void run() {
  while (!closed) {
    try {
      final BaseWorkerCmd command;
      // see rollback() for lock reason
      synchronized (cmdLock) {
        command = jobMaster.getRuntimeContext().foCmds.poll(1, TimeUnit.SECONDS);
      }
      if (null == command) {
        continue;
      }
      if (command instanceof WorkerRollbackRequest) {
        jobMaster.getRuntimeContext().unfinishedFoCmds.add(command);
        dealWithRollbackRequest((WorkerRollbackRequest) command);
      }
    } catch (Throwable e) {
      LOG.error("Fo coordinator occur err.", e);
    }
  }
  LOG.warn("Fo coordinator thread exit.");
}

private void dealWithRollbackRequest(WorkerRollbackRequest rollbackRequest) {
  LOG.info("Start deal with rollback request {}.", rollbackRequest);

  ExecutionVertex exeVertex = getExeVertexFromRequest(rollbackRequest);

  // Reset pid for new-rollback actor.
  if (null != rollbackRequest.getPid() &&
      !rollbackRequest.getPid().equals(WorkerRollbackRequest.DEFAULT_PID)) {
    exeVertex.setPid(rollbackRequest.getPid());
  }

  if (isRollbacking.get(exeVertex)) {
    LOG.info("Vertex {} is rollbacking, skip rollback again.", exeVertex);
    return;
  }

  String hostname = "";
  Optional<Container> container = ResourceUtil.getContainerById(
      jobMaster.getResourceManager().getRegisteredContainers(),
      exeVertex.getContainerId()
  );
  if (container.isPresent()) {
    hostname = container.get().getHostname();
  }

  if (rollbackRequest.isForcedRollback) {
    interruptCheckpointAndRollback(rollbackRequest);
  } else {
    asyncRemoteCaller.checkIfNeedRollbackAsync(exeVertex.getWorkerActor(), res -> {
      if (!res) {
        LOG.info("Vertex {} doesn't need to rollback, skip it.", exeVertex);
        return;
      }
      interruptCheckpointAndRollback(rollbackRequest);
    }, throwable -> {
      LOG.error("Exception when calling checkIfNeedRollbackAsync, maybe vertex is dead" +
          ", ignore this request, vertex={}.", exeVertex, throwable);
    });
  }

  LOG.info("Deal with rollback request {} success.", rollbackRequest);
}

private Boolean isDuplicateRequest(WorkerRollbackRequest request) {
  try {
    Object[] foCmdsArray = runtimeContext.foCmds.toArray();
    for (Object cmd : foCmdsArray) {
      if (request.fromActorId.equals(((BaseWorkerCmd) cmd).fromActorId)) {
        return true;
      }
    }
  } catch (Exception e) {
    LOG.warn("Check request is duplicated failed.", e);
  }
  return false;
}

private ExecutionVertex getExeVertexFromRequest(WorkerRollbackRequest rollbackRequest) {
  ActorId actorId = rollbackRequest.fromActorId;
  Optional<BaseActorHandle> rayActor = graphManager.getExecutionGraph().getActorById(actorId);
  if (!rayActor.isPresent()) {
    throw new RuntimeException("Can not find ray actor of ID " + actorId);
  }
  return getExecutionVertex(rollbackRequest.fromActorId);
}

private ExecutionVertex getExecutionVertex(BaseActorHandle actor) {
  return graphManager.getExecutionGraph().getExecutionVertexByActorId(actor.getId());
}

private ExecutionVertex getExecutionVertex(ActorId actorId) {
  return graphManager.getExecutionGraph().getExecutionVertexByActorId(actorId);
}

private void interruptCheckpointAndRollback(WorkerRollbackRequest rollbackRequest) {
  // assign a cascadingGroupId
  if (rollbackRequest.cascadingGroupId == null) {
    rollbackRequest.cascadingGroupId = currentCascadingGroupId++;
  }
  // get last valid checkpoint id then call worker rollback
  rollback(jobMaster.getRuntimeContext().getLastValidCheckpointId(), rollbackRequest,
      currentCascadingGroupId);
  // we interrupt current checkpoint for 2 considerations:
  // 1. current checkpoint might be timeout, because barrier might be lost after failover. so we
  // interrupt current checkpoint to avoid waiting.
  // 2. when we want to rollback vertex to n, job finished checkpoint n+1 and cleared state
  // of checkpoint n.
  jobMaster.getRuntimeContext().cpCmds.offer(new InterruptCheckpointRequest());
}

private void rollback(
    long checkpointId, WorkerRollbackRequest rollbackRequest,
    long cascadingGroupId) {
  ExecutionVertex exeVertex = getExeVertexFromRequest(rollbackRequest);
  LOG.info("Call vertex {} to rollback, checkpoint id is {}, cascadingGroupId={}.",
      exeVertex, checkpointId, cascadingGroupId);

  isRollbacking.put(exeVertex, true);

  asyncRemoteCaller.rollback(exeVertex.getWorkerActor(), checkpointId, result -> {
    List<WorkerRollbackRequest> newRollbackRequests = new ArrayList<>();
    switch (result.getResultEnum()) {
      case SUCCESS:
        ChannelRecoverInfo recoverInfo = result.getResultObj();
        LOG.info("Vertex {} rollback done, dataLostQueues={}, msg={}, cascadingGroupId={}.",
            exeVertex, recoverInfo.getDataLostQueues(), result.getResultMsg(), cascadingGroupId);
        // rollback upstream if vertex reports abnormal input queues
        newRollbackRequests =
            cascadeUpstreamActors(recoverInfo.getDataLostQueues(), exeVertex, cascadingGroupId);
        break;
      case SKIPPED:
        LOG.info("Vertex skip rollback, result = {}, cascadingGroupId={}.", result,
            cascadingGroupId);
        break;
      default:
        LOG.error(
            "Rollback vertex {} failed, result={}, cascadingGroupId={}," +
                " rollback this worker again after {} ms.",
            exeVertex, result, cascadingGroupId, ROLLBACK_RETRY_TIME_MS);
        Thread.sleep(ROLLBACK_RETRY_TIME_MS);
        LOG.info("Add rollback request for {} again, cascadingGroupId={}.", exeVertex,
            cascadingGroupId);
        newRollbackRequests.add(
            new WorkerRollbackRequest(exeVertex, "", "Rollback failed, try again.", false)
        );
        break;
    }

    // lock to avoid executing new rollback requests added.
    // consider such a case: A->B->C, C cascade B, and B cascade A
    // if B is rollback before B's rollback request is saved, and then JobMaster crashed,
    // then A will never be rollback.
    synchronized (cmdLock) {
      jobMaster.getRuntimeContext().foCmds.addAll(newRollbackRequests);
      // this rollback request is finished, remove it.
      jobMaster.getRuntimeContext().unfinishedFoCmds.remove(rollbackRequest);
      jobMaster.saveContext();
    }
    isRollbacking.put(exeVertex, false);
  }, throwable -> {
    LOG.error("Exception when calling vertex to rollback, vertex={}.", exeVertex, throwable);
    isRollbacking.put(exeVertex, false);
  });

  LOG.info("Finish rollback vertex {}, checkpoint id is {}.", exeVertex, checkpointId);
}

private List<WorkerRollbackRequest> cascadeUpstreamActors(
    Set<String> dataLostQueues, ExecutionVertex fromVertex, long cascadingGroupId) {
  List<WorkerRollbackRequest> cascadedRollbackRequest = new ArrayList<>();
  // rollback upstream if vertex reports abnormal input queues
  dataLostQueues.forEach(q -> {
    BaseActorHandle upstreamActor =
        graphManager.getExecutionGraph().getPeerActor(fromVertex.getWorkerActor(), q);
    ExecutionVertex upstreamExeVertex = getExecutionVertex(upstreamActor);
    // vertexes that has already cascaded by other vertex in the same level
    // of graph should be ignored.
    if (isRollbacking.get(upstreamExeVertex)) {
      return;
    }
    LOG.info("Call upstream vertex {} of vertex {} to rollback, cascadingGroupId={}.",
        upstreamExeVertex, fromVertex, cascadingGroupId);
    String hostname = "";
    Optional<Container> container = ResourceUtil.getContainerById(
        jobMaster.getResourceManager().getRegisteredContainers(),
        upstreamExeVertex.getContainerId()
    );
    if (container.isPresent()) {
      hostname = container.get().getHostname();
    }
    // force upstream vertexes to rollback
    WorkerRollbackRequest upstreamRequest = new WorkerRollbackRequest(
        upstreamExeVertex, hostname, String.format("Cascading rollback from %s", fromVertex), true
    );
    upstreamRequest.cascadingGroupId = cascadingGroupId;
    cascadedRollbackRequest.add(upstreamRequest);
  });
  return cascadedRollbackRequest;
}