This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 149785a5725 [FLINK-29488][metrics] MetricRegistryImpl implements
AutoCloseableAsync
149785a5725 is described below
commit 149785a5725a4727287d2f133b3b3abbb23f99a9
Author: Chesnay Schepler <[email protected]>
AuthorDate: Mon Sep 19 10:04:59 2022 +0200
[FLINK-29488][metrics] MetricRegistryImpl implements AutoCloseableAsync
---
.../flink/runtime/entrypoint/ClusterEntrypoint.java | 2 +-
.../flink/runtime/metrics/MetricRegistryImpl.java | 6 ++++--
.../apache/flink/runtime/minicluster/MiniCluster.java | 2 +-
.../flink/runtime/taskexecutor/TaskManagerRunner.java | 2 +-
.../flink/runtime/metrics/MetricRegistryImplTest.java | 18 +++++++++---------
.../metrics/groups/AbstractMetricGroupTest.java | 8 ++++----
.../metrics/groups/InternalOperatorGroupTest.java | 4 ++--
.../runtime/metrics/groups/JobManagerGroupTest.java | 8 ++++----
.../runtime/metrics/groups/JobManagerJobGroupTest.java | 6 +++---
.../metrics/groups/MetricGroupRegistrationTest.java | 4 ++--
.../flink/runtime/metrics/groups/MetricGroupTest.java | 4 ++--
.../runtime/metrics/groups/TaskManagerGroupTest.java | 4 ++--
.../metrics/groups/TaskManagerJobGroupTest.java | 6 +++---
.../runtime/metrics/groups/TaskMetricGroupTest.java | 10 +++++-----
.../TaskExecutorOperatorEventHandlingTest.java | 2 +-
15 files changed, 44 insertions(+), 42 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
index db14c3396b7..f0379f66d9c 100755
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
@@ -512,7 +512,7 @@ public abstract class ClusterEntrypoint implements
AutoCloseableAsync, FatalErro
}
if (metricRegistry != null) {
- terminationFutures.add(metricRegistry.shutdown());
+ terminationFutures.add(metricRegistry.closeAsync());
}
if (ioExecutor != null) {
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryImpl.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryImpl.java
index 8d26119ef58..d053ba58f72 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryImpl.java
@@ -35,6 +35,7 @@ import
org.apache.flink.runtime.metrics.groups.ReporterScopedSettings;
import org.apache.flink.runtime.metrics.scope.ScopeFormats;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceGateway;
+import org.apache.flink.util.AutoCloseableAsync;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.FlinkException;
@@ -67,7 +68,7 @@ import java.util.stream.Collectors;
* A MetricRegistry keeps track of all registered {@link Metric Metrics}. It
serves as the
* connection between {@link MetricGroup MetricGroups} and {@link
MetricReporter MetricReporters}.
*/
-public class MetricRegistryImpl implements MetricRegistry {
+public class MetricRegistryImpl implements MetricRegistry, AutoCloseableAsync {
private static final Logger LOG =
LoggerFactory.getLogger(MetricRegistryImpl.class);
private final Object lock = new Object();
@@ -312,7 +313,8 @@ public class MetricRegistryImpl implements MetricRegistry {
*
* @return Future which is completed once the {@link MetricRegistryImpl}
is shut down.
*/
- public CompletableFuture<Void> shutdown() {
+ @Override
+ public CompletableFuture<Void> closeAsync() {
synchronized (lock) {
if (isShutdown) {
return terminationFuture;
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
index 3a11fa93473..3f447bb71e0 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
@@ -706,7 +706,7 @@ public class MiniCluster implements AutoCloseableAsync {
// metrics shutdown
if (metricRegistry != null) {
- terminationFutures.add(metricRegistry.shutdown());
+ terminationFutures.add(metricRegistry.closeAsync());
metricRegistry = null;
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
index 0c2b82ff2b5..f602c810f5d 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -380,7 +380,7 @@ public class TaskManagerRunner implements FatalErrorHandler
{
if (metricRegistry != null) {
try {
- metricRegistry.shutdown();
+ metricRegistry.closeAsync();
} catch (Exception e) {
exception = ExceptionUtils.firstOrSuppressed(e, exception);
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/MetricRegistryImplTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/MetricRegistryImplTest.java
index 8748c39681a..255b4a35493 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/MetricRegistryImplTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/MetricRegistryImplTest.java
@@ -77,7 +77,7 @@ class MetricRegistryImplTest {
assertThat(metricRegistry.isShutdown()).isFalse();
- metricRegistry.shutdown().get();
+ metricRegistry.closeAsync().get();
assertThat(metricRegistry.isShutdown()).isTrue();
}
@@ -161,7 +161,7 @@ class MetricRegistryImplTest {
}
assertThat(reporter.getReportCount()).as("No report was
triggered.").isGreaterThan(0);
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
@@ -188,7 +188,7 @@ class MetricRegistryImplTest {
assertThat(reportTask.getDelay(TimeUnit.SECONDS))
.isEqualTo(MetricOptions.REPORTER_INTERVAL.defaultValue().getSeconds());
} finally {
- registry.shutdown().get();
+ registry.closeAsync().get();
}
}
@@ -242,7 +242,7 @@ class MetricRegistryImplTest {
assertThat(reporter2.getLastRemovedMetric()).containsInstanceOf(Counter.class);
assertThat(reporter2.getLastRemovedMetricName()).hasValue("rootCounter");
- registry.shutdown().get();
+ registry.closeAsync().get();
}
/**
@@ -319,7 +319,7 @@ class MetricRegistryImplTest {
registry, "host", new ResourceID("id"));
assertThat(tmGroup.getMetricIdentifier("name")).isEqualTo("A_B_C_D_E_name");
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
@@ -348,7 +348,7 @@ class MetricRegistryImplTest {
assertThat(registry.getDelimiter(3)).isEqualTo(GLOBAL_DEFAULT_DELIMITER);
assertThat(registry.getDelimiter(-1)).isEqualTo(GLOBAL_DEFAULT_DELIMITER);
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
@@ -384,7 +384,7 @@ class MetricRegistryImplTest {
registry, "host", new ResourceID("id"));
group.counter(name);
group.close();
- registry.shutdown().get();
+ registry.closeAsync().get();
for (ReporterSetup cfg : reporterConfigurations) {
String delimiter =
@@ -421,7 +421,7 @@ class MetricRegistryImplTest {
MetricQueryService queryService =
checkNotNull(registry.getQueryService());
- registry.shutdown().get();
+ registry.closeAsync().get();
queryService.getTerminationFuture().get();
}
@@ -450,7 +450,7 @@ class MetricRegistryImplTest {
assertThat(reporter1.getLastRemovedMetric()).hasValue(metric);
assertThat(reporter1.getLastRemovedMetricName()).hasValue("counter");
- registry.shutdown().get();
+ registry.closeAsync().get();
}
/** Reporter that throws an exception when it is notified of an added or
removed metric. */
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java
index a672336b682..c2ffe198ced 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java
@@ -77,7 +77,7 @@ public class AbstractMetricGroupTest extends TestLogger {
};
assertTrue(group.getAllVariables().isEmpty());
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
@@ -237,7 +237,7 @@ public class AbstractMetricGroupTest extends TestLogger {
input -> input.replace("A",
"X").replace(counterName, "3")));
}
} finally {
- testRegistry.shutdown().get();
+ testRegistry.closeAsync().get();
}
}
@@ -275,7 +275,7 @@ public class AbstractMetricGroupTest extends TestLogger {
reporter2.findAdded(counterName).group)
.getLogicalScope(reporter2, ','));
} finally {
- testRegistry.shutdown().get();
+ testRegistry.closeAsync().get();
}
}
@@ -301,7 +301,7 @@ public class AbstractMetricGroupTest extends TestLogger {
assertEquals("A.X.C.D.1", group.getMetricIdentifier("1", FILTER_B,
-1, '.'));
assertEquals("A.X.C.D.1", group.getMetricIdentifier("1", FILTER_B,
2, '.'));
} finally {
- testRegistry.shutdown().get();
+ testRegistry.closeAsync().get();
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/InternalOperatorGroupTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/InternalOperatorGroupTest.java
index 50d307eaad4..487421d627d 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/InternalOperatorGroupTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/InternalOperatorGroupTest.java
@@ -58,7 +58,7 @@ public class InternalOperatorGroupTest extends TestLogger {
@After
public void teardown() throws Exception {
if (registry != null) {
- registry.shutdown().get();
+ registry.closeAsync().get();
}
}
@@ -122,7 +122,7 @@ public class InternalOperatorGroupTest extends TestLogger {
"%s.%s.%s.%s.%s.name", tmID, jid, vertexId,
operatorName, operatorID),
operatorGroup.getMetricIdentifier("name"));
} finally {
- registry.shutdown().get();
+ registry.closeAsync().get();
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/JobManagerGroupTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/JobManagerGroupTest.java
index 4a06fa6d8c8..90a8d41a4b5 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/JobManagerGroupTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/JobManagerGroupTest.java
@@ -73,7 +73,7 @@ public class JobManagerGroupTest extends TestLogger {
assertTrue(jmJobGroup21.isClosed());
assertEquals(0, group.numRegisteredJobMetricGroups());
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
@@ -98,7 +98,7 @@ public class JobManagerGroupTest extends TestLogger {
assertTrue(jmJobGroup11.isClosed());
assertTrue(jmJobGroup21.isClosed());
- registry.shutdown().get();
+ registry.closeAsync().get();
}
// ------------------------------------------------------------------------
@@ -116,7 +116,7 @@ public class JobManagerGroupTest extends TestLogger {
assertArrayEquals(new String[] {"localhost", "jobmanager"},
group.getScopeComponents());
assertEquals("localhost.jobmanager.name",
group.getMetricIdentifier("name"));
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
@@ -133,7 +133,7 @@ public class JobManagerGroupTest extends TestLogger {
new String[] {"constant", "host", "foo", "host"},
group.getScopeComponents());
assertEquals("constant.host.foo.host.name",
group.getMetricIdentifier("name"));
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/JobManagerJobGroupTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/JobManagerJobGroupTest.java
index 084c3c6f6ac..f24cae5519b 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/JobManagerJobGroupTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/JobManagerJobGroupTest.java
@@ -51,7 +51,7 @@ public class JobManagerJobGroupTest extends TestLogger {
assertEquals("theHostName.jobmanager.myJobName.name",
jmGroup.getMetricIdentifier("name"));
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
@@ -71,7 +71,7 @@ public class JobManagerJobGroupTest extends TestLogger {
assertEquals("some-constant.myJobName.name",
jmGroup.getMetricIdentifier("name"));
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
@@ -94,7 +94,7 @@ public class JobManagerJobGroupTest extends TestLogger {
assertEquals("peter.some-constant." + jid + ".name",
jmGroup.getMetricIdentifier("name"));
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupRegistrationTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupRegistrationTest.java
index 47041b4a23f..f52a00e4607 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupRegistrationTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupRegistrationTest.java
@@ -91,7 +91,7 @@ public class MetricGroupRegistrationTest extends TestLogger {
Assert.assertEquals(histogram, TestReporter1.lastPassedMetric);
assertEquals("histogram", TestReporter1.lastPassedName);
- registry.shutdown().get();
+ registry.closeAsync().get();
}
/** Reporter that exposes the last name and metric instance it was
notified of. */
@@ -127,6 +127,6 @@ public class MetricGroupRegistrationTest extends TestLogger
{
MetricGroup group3 = root.addGroup("group");
Assert.assertTrue(group1 == group2 && group2 == group3);
- registry.shutdown().get();
+ registry.closeAsync().get();
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java
index 821a02473cc..b9d93419be8 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java
@@ -73,7 +73,7 @@ public class MetricGroupTest extends TestLogger {
@After
public void shutdownRegistry() throws Exception {
- this.registry.shutdown().get();
+ this.registry.closeAsync().get();
this.registry = null;
}
@@ -266,7 +266,7 @@ public class MetricGroupTest extends TestLogger {
assertThat(
"Value is present in logical scope.", logicalScope,
not(containsString(value)));
} finally {
- registry.shutdown().get();
+ registry.closeAsync().get();
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/TaskManagerGroupTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/TaskManagerGroupTest.java
index 948aa9f05ae..d055c142154 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/TaskManagerGroupTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/TaskManagerGroupTest.java
@@ -59,7 +59,7 @@ public class TaskManagerGroupTest extends TestLogger {
@After
public void teardown() throws Exception {
if (registry != null) {
- registry.shutdown().get();
+ registry.closeAsync().get();
}
}
@@ -185,7 +185,7 @@ public class TaskManagerGroupTest extends TestLogger {
assertArrayEquals(
new String[] {"constant", "host", "foo", "host"},
group.getScopeComponents());
assertEquals("constant.host.foo.host.name",
group.getMetricIdentifier("name"));
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/TaskManagerJobGroupTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/TaskManagerJobGroupTest.java
index c46140a00ae..f2df4813c98 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/TaskManagerJobGroupTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/TaskManagerJobGroupTest.java
@@ -50,7 +50,7 @@ public class TaskManagerJobGroupTest extends TestLogger {
@After
public void teardown() throws Exception {
if (registry != null) {
- registry.shutdown().get();
+ registry.closeAsync().get();
}
}
@@ -90,7 +90,7 @@ public class TaskManagerJobGroupTest extends TestLogger {
new String[] {"some-constant", "myJobName"},
jmGroup.getScopeComponents());
assertEquals("some-constant.myJobName.name",
jmGroup.getMetricIdentifier("name"));
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
@@ -115,7 +115,7 @@ public class TaskManagerJobGroupTest extends TestLogger {
assertEquals(
"peter.test-tm-id.some-constant." + jid + ".name",
jmGroup.getMetricIdentifier("name"));
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroupTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroupTest.java
index c732afb3242..46d45f4e08a 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroupTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroupTest.java
@@ -57,7 +57,7 @@ public class TaskMetricGroupTest extends TestLogger {
@After
public void teardown() throws Exception {
if (registry != null) {
- registry.shutdown().get();
+ registry.closeAsync().get();
}
}
@@ -116,7 +116,7 @@ public class TaskMetricGroupTest extends TestLogger {
assertEquals(
String.format("test-tm-id.%s.%s.%s.name", jid, vertexId,
executionId),
taskGroup.getMetricIdentifier("name"));
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
@@ -149,7 +149,7 @@ public class TaskMetricGroupTest extends TestLogger {
assertEquals(
"theHostName.taskmanager.test-tm-id.myJobName." + executionId
+ ".13.name",
taskGroup.getMetricIdentifier("name"));
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
@@ -192,7 +192,7 @@ public class TaskMetricGroupTest extends TestLogger {
// now all registered metrics should have been unregistered
assertEquals(initialMetricsCount,
registry.getNumberRegisteredMetrics());
- registry.shutdown().get();
+ registry.closeAsync().get();
}
@Test
@@ -217,7 +217,7 @@ public class TaskMetricGroupTest extends TestLogger {
Assert.assertEquals(
originalName.substring(0,
TaskMetricGroup.METRICS_OPERATOR_NAME_MAX_LENGTH),
storedName);
- registry.shutdown().get();
+ registry.closeAsync().get();
}
private static class CountingMetricRegistry extends MetricRegistryImpl {
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorOperatorEventHandlingTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorOperatorEventHandlingTest.java
index 9248071bc5d..c6b6013fb40 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorOperatorEventHandlingTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorOperatorEventHandlingTest.java
@@ -90,7 +90,7 @@ public class TaskExecutorOperatorEventHandlingTest extends
TestLogger {
}
if (metricRegistry != null) {
- metricRegistry.shutdown().get();
+ metricRegistry.closeAsync().get();
}
}
