code: private void handleTaskException(final TaskClientCodeException e) {
  
      LOG.log(Level.SEVERE, "TaskClientCodeException", e);
  
      final ByteString exception = ByteString.copyFrom(this.exceptionCodec.toBytes(e.getCause()));
  
      final ReefServiceProtos.TaskStatusProto taskStatus =
          ReefServiceProtos.TaskStatusProto.newBuilder()
              .setContextId(e.getContextId())
              .setTaskId(e.getTaskId())
              .setResult(exception)
              .setState(ReefServiceProtos.State.FAILED)
              .build();
  
      LOG.log(Level.SEVERE, "Sending heartbeat: {0}", taskStatus);
  
      this.heartBeatManager.sendTaskStatus(taskStatus);
    }
docs: THIS ASSUMES THAT IT IS CALLED ON A THREAD HOLDING THE LOCK ON THE HeartBeatManager.