[FLINK-3251] [runtime] Return empty stats for unknown operator

nltran · Jan 18, 2016 · 117ba95 · 117ba95
1 parent 1c3bbe4
commit 117ba95
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 24 deletions.
diff --git a/...src/main/java/org/apache/flink/runtime/checkpoint/stats/SimpleCheckpointStatsTracker.java b/...src/main/java/org/apache/flink/runtime/checkpoint/stats/SimpleCheckpointStatsTracker.java
@@ -299,31 +299,32 @@ else if (latestCompletedCheckpoint != null && subTaskStats != null) {
  long[][] subTaskStats = this.subTaskStats.get(operatorId);
 
  if (subTaskStats == null) {
- throw new IllegalArgumentException("Unknown operator ID.");
+ return Option.empty();
  }
+ else {
+ long maxDuration = Long.MIN_VALUE;
+ long stateSize = 0;
 
- long maxDuration = Long.MIN_VALUE;
- long stateSize = 0;
+ for (long[] subTaskStat : subTaskStats) {
+ if (subTaskStat[0] > maxDuration) {
+ maxDuration = subTaskStat[0];
+ }
 
- for (long[] subTaskStat : subTaskStats) {
- if (subTaskStat[0] > maxDuration) {
- maxDuration = subTaskStat[0];
+ stateSize += subTaskStat[1];
  }
 
- stateSize += subTaskStat[1];
- }
-
- stats = new OperatorCheckpointStats(
- latestCompletedCheckpoint.getCheckpointID(),
- latestCompletedCheckpoint.getTimestamp(),
- maxDuration,
- stateSize,
- subTaskStats);
+ stats = new OperatorCheckpointStats(
+ latestCompletedCheckpoint.getCheckpointID(),
+ latestCompletedCheckpoint.getTimestamp(),
+ maxDuration,
+ stateSize,
+ subTaskStats);
 
- // Remember this and don't recompute if requested again
- operatorStatsCache.put(operatorId, stats);
+  // Remember this and don't recompute if requested again
+  operatorStatsCache.put(operatorId, stats);
 
- return Option.apply(stats);
+ return Option.apply(stats);
+ }
  }
  else {
  return Option.empty();

diff --git a/...test/java/org/apache/flink/runtime/checkpoint/stats/SimpleCheckpointStatsTrackerTest.java b/...test/java/org/apache/flink/runtime/checkpoint/stats/SimpleCheckpointStatsTrackerTest.java
@@ -85,12 +85,7 @@ public void testIllegalOperatorId() throws Exception {
 
  assertTrue(tracker.getJobStats().isDefined());
 
- try {
- tracker.getOperatorStats(new JobVertexID());
- fail("Did not throw expected Exception");
- }
- catch (IllegalArgumentException ignored) {
- }
+ assertTrue(tracker.getOperatorStats(new JobVertexID()).isEmpty());
  }
 
  @Test