Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5184#discussion_r160407686
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
---
@@ -956,23 +958,29 @@ private void jobStatusChanged(
"The job is
registered as 'FINISHED (successful), but this notification describes " +
"a failure,
since the resulting accumulators could not be fetched.", e);
- executor.execute(()
->jobCompletionActions.jobFailed(exception));
+ executor.execute(() ->
jobCompletionActions.jobFailed(builder
+
.serializedThrowable(new SerializedThrowable(exception))
+ .build()));
}
break;
case CANCELED: {
final JobExecutionException exception =
new JobExecutionException(
jobID, "Job was cancelled.",
new Exception("The job was cancelled"));
- executor.execute(() ->
jobCompletionActions.jobFailed(exception));
+ executor.execute(() ->
jobCompletionActions.jobFailed(builder
+ .serializedThrowable(new
SerializedThrowable(exception))
+ .build()));
break;
}
case FAILED: {
final Throwable unpackedError =
SerializedThrowable.get(error, userCodeLoader);
--- End diff --
I think `error` is always a `SerializedThrowable`.
```
private void notifyJobStatusChange(JobStatus newState, Throwable error)
{
if (jobStatusListeners.size() > 0) {
final long timestamp = System.currentTimeMillis();
final Throwable serializedError = error == null ? null
: new SerializedThrowable(error);
for (JobStatusListener listener : jobStatusListeners) {
try {
listener.jobStatusChanges(getJobID(),
newState, timestamp, serializedError);
} catch (Throwable t) {
LOG.warn("Error while notifying
JobStatusListener", t);
}
}
}
}
```
Not sure if it makes sense to unpack, wrap, and serialize it again.
---
