[FLINK-6183] [metrics] Prevent some cases of TaskMG not being closed

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerJobMetricGroup.java

@@ -72,16 +72,21 @@ public class TaskManagerJobMetricGroup extends JobMetricGroup<TaskManagerMetricG
 
 		synchronized (this) {
 			if (!isClosed()) {
-				TaskMetricGroup task = new TaskMetricGroup(
-					registry,
-					this,
-					jobVertexId,
-					executionAttemptID,
-					taskName,
-					subtaskIndex,
-					attemptNumber);
-				tasks.put(executionAttemptID, task);
-				return task;
+				TaskMetricGroup prior = tasks.get(executionAttemptID);
+				if (prior != null) {
+					return prior;
+				} else {
+					TaskMetricGroup task = new TaskMetricGroup(
+						registry,
+						this,
+						jobVertexId,
+						executionAttemptID,
+						taskName,
+						subtaskIndex,
+						attemptNumber);
+					tasks.put(executionAttemptID, task);
+					return task;
+				}
 			} else {
 				return null;
 			}

flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java

@@ -526,16 +526,25 @@ public class Task implements Runnable, TaskActions {
 			else if (current == ExecutionState.FAILED) {
 				// we were immediately failed. tell the TaskManager that we reached our final state
 				notifyFinalState();
+				if (metrics != null) {
+					metrics.close();
+				}
 				return;
 			}
 			else if (current == ExecutionState.CANCELING) {
 				if (transitionState(ExecutionState.CANCELING, ExecutionState.CANCELED)) {
 					// we were immediately canceled. tell the TaskManager that we reached our final state
 					notifyFinalState();
+					if (metrics != null) {
+						metrics.close();
+					}
 					return;
 				}
 			}
 			else {
+				if (metrics != null) {
+					metrics.close();
+				}
 				throw new IllegalStateException("Invalid state for beginning of operation of task " + this + '.');
 			}
 		}