MasterService.java 文件源码

/**
 * worker run over successfully
 *
 * @param controller rpc controller of protobuf
 * @param request contains worker attempt id
 * @throws ServiceException
 */
@SuppressWarnings("unchecked")
@Override
public WorkerDoneResponse workerDone(RpcController controller, WorkerDoneRequest request)
    throws ServiceException {
  WorkerAttemptId workerAttemptId = ProtobufUtil.convertToId(request.getWorkerAttemptId());
  LOG.info("worker attempt " + workerAttemptId + " is done");
  WorkerDoneResponse.Builder resBuilder = WorkerDoneResponse.newBuilder();

  //if worker attempt id is not in monitor set, we should shutdown it
  if (!workerLastHeartbeatTS.containsKey(workerAttemptId)) {
    resBuilder.setCommand(WorkerCommandProto.W_SHUTDOWN);
  } else {
    workerLastHeartbeatTS.remove(workerAttemptId);
    resBuilder.setCommand(WorkerCommandProto.W_SUCCESS);
    context.getEventHandler().handle(new WorkerAttemptEvent(WorkerAttemptEventType.DONE, workerAttemptId));
  }

  return resBuilder.build();
}