[FLINK-12247][rest] Fix NPE when writing the archive json file to Fil…

…eSystem This closes apache#8250.
bbejeck · Apr 24, 2019 · 191b9df · 191b9df
1 parent 8a17483
commit 191b9df
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 14 deletions.
diff --git a/...org/apache/flink/runtime/rest/handler/job/SubtaskExecutionAttemptAccumulatorsHandler.java b/...org/apache/flink/runtime/rest/handler/job/SubtaskExecutionAttemptAccumulatorsHandler.java
@@ -100,13 +100,15 @@ public Collection<ArchivedJson> archiveJsonWithPath(AccessExecutionGraph graph)
 
  for (int x = 0; x < subtask.getCurrentExecutionAttempt().getAttemptNumber(); x++) {
  AccessExecution attempt = subtask.getPriorExecutionAttempt(x);
- ResponseBody json = createAccumulatorInfo(attempt);
- String path = getMessageHeaders().getTargetRestEndpointURL()
- .replace(':' + JobIDPathParameter.KEY, graph.getJobID().toString())
- .replace(':' + JobVertexIdPathParameter.KEY, task.getJobVertexId().toString())
- .replace(':' + SubtaskIndexPathParameter.KEY, String.valueOf(subtask.getParallelSubtaskIndex()))
- .replace(':' + SubtaskAttemptPathParameter.KEY, String.valueOf(attempt.getAttemptNumber()));
- archive.add(new ArchivedJson(path, json));
+ if (attempt != null){
+ ResponseBody json = createAccumulatorInfo(attempt);
+ String path = getMessageHeaders().getTargetRestEndpointURL()
+ .replace(':' + JobIDPathParameter.KEY, graph.getJobID().toString())
+ .replace(':' + JobVertexIdPathParameter.KEY, task.getJobVertexId().toString())
+ .replace(':' + SubtaskIndexPathParameter.KEY, String.valueOf(subtask.getParallelSubtaskIndex()))
+ .replace(':' + SubtaskAttemptPathParameter.KEY, String.valueOf(attempt.getAttemptNumber()));
+ archive.add(new ArchivedJson(path, json));
+ }
  }
  }
  }

diff --git a/...java/org/apache/flink/runtime/rest/handler/job/SubtaskExecutionAttemptDetailsHandler.java b/...java/org/apache/flink/runtime/rest/handler/job/SubtaskExecutionAttemptDetailsHandler.java
@@ -114,13 +114,15 @@ public Collection<ArchivedJson> archiveJsonWithPath(AccessExecutionGraph graph)
 
  for (int x = 0; x < subtask.getCurrentExecutionAttempt().getAttemptNumber(); x++) {
  AccessExecution attempt = subtask.getPriorExecutionAttempt(x);
- ResponseBody json = createDetailsInfo(attempt, graph.getJobID(), task.getJobVertexId(), null);
- String path = getMessageHeaders().getTargetRestEndpointURL()
- .replace(':' + JobIDPathParameter.KEY, graph.getJobID().toString())
- .replace(':' + JobVertexIdPathParameter.KEY, task.getJobVertexId().toString())
- .replace(':' + SubtaskIndexPathParameter.KEY, String.valueOf(subtask.getParallelSubtaskIndex()))
- .replace(':' + SubtaskAttemptPathParameter.KEY, String.valueOf(attempt.getAttemptNumber()));
- archive.add(new ArchivedJson(path, json));
+ if (attempt != null) {
+ ResponseBody json = createDetailsInfo(attempt, graph.getJobID(), task.getJobVertexId(), null);
+ String path = getMessageHeaders().getTargetRestEndpointURL()
+ .replace(':' + JobIDPathParameter.KEY, graph.getJobID().toString())
+ .replace(':' + JobVertexIdPathParameter.KEY, task.getJobVertexId().toString())
+ .replace(':' + SubtaskIndexPathParameter.KEY, String.valueOf(subtask.getParallelSubtaskIndex()))
+ .replace(':' + SubtaskAttemptPathParameter.KEY, String.valueOf(attempt.getAttemptNumber()));
+ archive.add(new ArchivedJson(path, json));
+ }
  }
  }
  }