[FLINK-4210][metrics] Move close()/isClosed() out of MetricGroup

This closes #2286

flink-core/src/main/java/org/apache/flink/metrics/MetricGroup.java

@@ -27,34 +27,10 @@ import org.apache.flink.annotation.PublicEvolving;
  * hierarchy based on the group names.
  * 
  * <p>A MetricGroup is uniquely identified by it's place in the hierarchy and name.
- * 
- * <p>Metrics groups can be {@link #close() closed}. Upon closing, the group de-register all metrics
- * from any metrics reporter and any internal maps. Note that even closed metrics groups
- * return Counters, Gauges, etc to the code, to prevent exceptions in the monitored code.
- * These metrics simply do not get reported any more, when created on a closed group.
  */
 @PublicEvolving
 public interface MetricGroup {
 
-	// ------------------------------------------------------------------------
-	//  Closing
-	// ------------------------------------------------------------------------
-
-	/**
-	 * Marks the group as closed.
-	 * Recursively unregisters all {@link Metric Metrics} contained in this group.
-	 * 
-	 * <p>Any metrics created after the call to this function will not be registered in
-	 * the {@link MetricRegistry} and not be reported to any reporter (like JMX).
-	 */
-	void close();
-
-	/**
-	 * Checks whether this MetricGroup has been closed. 
-	 * @return True if the group has been closed, false is the group is still open.
-	 */
-	boolean isClosed();
-
 	// ------------------------------------------------------------------------
 	//  Metrics
 	// ------------------------------------------------------------------------

flink-core/src/main/java/org/apache/flink/metrics/groups/AbstractMetricGroup.java

@@ -49,6 +49,11 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
  * be strictly from parent group to subgroup. If at any point, a subgroup holds its group
  * lock and calls a parent method that also acquires the lock, it will create a deadlock
  * condition.
+ *
+ * <p>An AbstractMetricGroup can be {@link #close() closed}. Upon closing, the group de-register all metrics
+ * from any metrics reporter and any internal maps. Note that even closed metrics groups
+ * return Counters, Gauges, etc to the code, to prevent exceptions in the monitored code.
+ * These metrics simply do not get reported any more, when created on a closed group.
  */
 @Internal
 public abstract class AbstractMetricGroup implements MetricGroup {
@@ -65,7 +70,7 @@ public abstract class AbstractMetricGroup implements MetricGroup {
 	private final Map<String, Metric> metrics = new HashMap<>();
 
 	/** All metric subgroups of this group */
-	private final Map<String, MetricGroup> groups = new HashMap<>();
+	private final Map<String, AbstractMetricGroup> groups = new HashMap<>();
 
 	/** The metrics scope represented by this group.
 	 *  For example ["host-7", "taskmanager-2", "window_word_count", "my-mapper" ]. */
@@ -132,14 +137,13 @@ public abstract class AbstractMetricGroup implements MetricGroup {
 	//  Closing
 	// ------------------------------------------------------------------------
 
-	@Override
 	public void close() {
 		synchronized (this) {
 			if (!closed) {
 				closed = true;
 
 				// close all subgroups
-				for (MetricGroup group : groups.values()) {
+				for (AbstractMetricGroup group : groups.values()) {
 					group.close();
 				}
 				groups.clear();
@@ -153,7 +157,6 @@ public abstract class AbstractMetricGroup implements MetricGroup {
 		}
 	}
 
-	@Override
 	public final boolean isClosed() {
 		return closed;
 	}
@@ -267,8 +270,8 @@ public abstract class AbstractMetricGroup implements MetricGroup {
 							name + "'. Metric might not get properly reported. (" + scopeString + ')');
 				}
 
-				MetricGroup newGroup = new GenericMetricGroup(registry, this, name);
-				MetricGroup prior = groups.put(name, newGroup);
+				AbstractMetricGroup newGroup = new GenericMetricGroup(registry, this, name);
+				AbstractMetricGroup prior = groups.put(name, newGroup);
 				if (prior == null) {
 					// no prior group with that name
 					return newGroup;

flink-core/src/main/java/org/apache/flink/metrics/groups/ProxyMetricGroup.java

@@ -38,16 +38,6 @@ public class ProxyMetricGroup<P extends MetricGroup> implements MetricGroup {
 		this.parentMetricGroup = Preconditions.checkNotNull(parentMetricGroup);
 	}
 
-	@Override
-	public final void close() {
-		// don't close the parent metric group because it can also contain other metrics
-	}
-
-	@Override
-	public final boolean isClosed() {
-		return parentMetricGroup.isClosed();
-	}
-
 	@Override
 	public final Counter counter(int name) {
 		return parentMetricGroup.counter(name);

flink-core/src/main/java/org/apache/flink/metrics/groups/UnregisteredMetricsGroup.java

@@ -28,20 +28,10 @@ import org.apache.flink.metrics.SimpleCounter;
 /**
  * A special {@link MetricGroup} that does not register any metrics at the metrics registry
  * and any reporters.
- * 
- * <p>This metrics group appears always closed ({@link #isClosed()}).
  */
 @Internal
 public class UnregisteredMetricsGroup implements MetricGroup {
 
-	@Override
-	public void close() {}
-
-	@Override
-	public boolean isClosed() {
-		return true;
-	}
-
 	@Override
 	public Counter counter(int name) {
 		return new SimpleCounter();

flink-core/src/test/java/org/apache/flink/metrics/MetricRegistryTest.java

@@ -131,7 +131,7 @@ public class MetricRegistryTest extends TestLogger {
 
 		MetricRegistry registry = new MetricRegistry(config);
 
-		MetricGroup root = new TaskManagerMetricGroup(registry, "host", "id");
+		TaskManagerMetricGroup root = new TaskManagerMetricGroup(registry, "host", "id");
 		root.counter("rootCounter");
 		root.close();
 

flink-core/src/test/java/org/apache/flink/metrics/groups/MetricGroupTest.java

@@ -26,7 +26,6 @@ import org.apache.flink.metrics.MetricRegistry;
 
 import org.junit.After;
 import org.junit.Before;
-import org.junit.BeforeClass;
 import org.junit.Test;
 
 import static org.junit.Assert.*;
@@ -88,7 +87,7 @@ public class MetricGroupTest {
 		group.close();
 		assertTrue(group.isClosed());
 		
-		MetricGroup subgroup = group.addGroup("test subgroup");
+		AbstractMetricGroup subgroup = (AbstractMetricGroup) group.addGroup("test subgroup");
 		assertTrue(subgroup.isClosed());
 	}
 

flink-core/src/test/java/org/apache/flink/metrics/groups/TaskMetricGroupTest.java

@@ -131,15 +131,11 @@ public class TaskMetricGroupTest {
 		TaskManagerJobMetricGroup taskManagerJobMetricGroup = new TaskManagerJobMetricGroup(registry, taskManagerMetricGroup, new JobID(), "job");
 		TaskMetricGroup taskMetricGroup = new TaskMetricGroup(registry, taskManagerJobMetricGroup, new AbstractID(), new AbstractID(), "task", 0, 0);
 
-		IOMetricGroup ioMetricGroup = taskMetricGroup.getIOMetricGroup();
-
 		// the io metric should have registered predefined metrics
 		assertTrue(registry.getNumberRegisteredMetrics() > 0);
 
 		taskMetricGroup.close();
 
-		assertTrue(ioMetricGroup.isClosed());
-
 		// now alle registered metrics should have been unregistered
 		assertEquals(0, registry.getNumberRegisteredMetrics());
 	}