code: private void onTaskStatusMessage(final TaskStatusPOJO taskStatus) {
  
      if (!(this.task.isPresent() && this.task.get().getId().equals(taskStatus.getTaskId()))) {
  
        final State state = taskStatus.getState();
        if (state.isRestartable() ||
            this.driverRestartManager.getEvaluatorRestartState(this.evaluatorId).isReregistered()) {
  
          // [REEF-308] exposes a bug where the .NET evaluator does not send its states in the right order
          // [REEF-289] is a related item which may fix the issue
          if (state.isRunning()) {
            LOG.log(Level.WARNING,
                "Received a message of state {0} for Task {1} before receiving its {2} state",
                new Object[] {State.RUNNING, taskStatus.getTaskId(), State.INIT});
          }
  
          // FAILED is a legal first state of a Task as it could have failed during construction.
          this.task = Optional.of(
              new TaskRepresenter(taskStatus.getTaskId(),
                  this.contextRepresenters.getContext(taskStatus.getContextId()),
                  this.messageDispatcher,
                  this,
                  this.exceptionCodec,
                  this.driverRestartManager));
        } else {
          throw new RuntimeException("Received a message of state " + state +
              ", not INIT, RUNNING, or FAILED for Task " + taskStatus.getTaskId() +
              " which we haven't heard from before.");
        }
      }
  
      this.task.get().onTaskStatusMessage(taskStatus);
  
      if (this.task.get().isNotRunning()) {
        LOG.log(Level.FINEST, "Task no longer running. De-registering it.");
        this.task = Optional.empty();
      }
    }
docs: Handle task status messages.
  
  @param taskStatus message contains the current task status.