[
https://issues.apache.org/jira/browse/FLINK-8234?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16318407#comment-16318407
]
ASF GitHub Bot commented on FLINK-8234:
---------------------------------------
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.
> Cache JobExecutionResult from finished JobManagerRunners
> --------------------------------------------------------
>
> Key: FLINK-8234
> URL: https://issues.apache.org/jira/browse/FLINK-8234
> Project: Flink
> Issue Type: Sub-task
> Components: Distributed Coordination
> Affects Versions: 1.5.0
> Reporter: Till Rohrmann
> Assignee: Gary Yao
> Priority: Blocker
> Labels: flip-6
> Fix For: 1.5.0
>
>
> In order to serve the {{JobExecutionResults}} we have to cache them in the
> {{Dispatcher}} after the {{JobManagerRunner}} has finished. The cache should
> have a configurable size and should periodically clean up stale entries in
> order to avoid memory exhaustion.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
