code: private synchronized void onJobFailure(final JobStatusProto jobStatusProto) {
      assert jobStatusProto.getState() == ReefServiceProtos.State.FAILED;
  
      final String id = this.jobId;
      final Optional<byte[]> data = jobStatusProto.hasException() ?
          Optional.of(jobStatusProto.getException().toByteArray()) :
          Optional.<byte[]>empty();
      final Optional<Throwable> cause = this.exceptionCodec.fromBytes(data);
  
      final String message;
      if (cause.isPresent() && cause.get().getMessage() != null) {
        message = cause.get().getMessage();
      } else {
        message = "No Message sent by the Job in exception " + cause.get();
        LOG.log(Level.WARNING, message, cause.get());
      }
      final Optional<String> description = Optional.of(message);
  
      final FailedJob failedJob = new FailedJob(id, message, description, cause, data);
      this.failedJobEventHandler.onNext(failedJob);
    }
docs: Inform the client of a failed job.
  
  @param jobStatusProto status of the failed job