You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2018/12/11 13:48:06 UTC
[flink] 03/04: [hotfix][metrics][tests] Ensure ActorSystem is
terminated
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
commit 8492d006984626c3d0d4fadaa8433dfde6c17023
Author: zentol <ch...@apache.org>
AuthorDate: Fri Dec 7 15:58:34 2018 +0100
[hotfix][metrics][tests] Ensure ActorSystem is terminated
---
.../metrics/dump/MetricQueryServiceTest.java | 300 +++++++++++----------
1 file changed, 152 insertions(+), 148 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/dump/MetricQueryServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/dump/MetricQueryServiceTest.java
index ccc2236..5d538ed 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/dump/MetricQueryServiceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/dump/MetricQueryServiceTest.java
@@ -53,173 +53,177 @@ public class MetricQueryServiceTest extends TestLogger {
@Test
public void testCreateDump() throws Exception {
ActorSystem s = AkkaUtils.createLocalActorSystem(new Configuration());
- ActorRef serviceActor = MetricQueryService.startMetricQueryService(s, null, Long.MAX_VALUE);
- TestActorRef testActorRef = TestActorRef.create(s, Props.create(TestActor.class));
- TestActor testActor = (TestActor) testActorRef.underlyingActor();
-
- final Counter c = new SimpleCounter();
- final Gauge<String> g = new Gauge<String>() {
- @Override
- public String getValue() {
- return "Hello";
+ try {
+ ActorRef serviceActor = MetricQueryService.startMetricQueryService(s, null, Long.MAX_VALUE);
+ TestActorRef testActorRef = TestActorRef.create(s, Props.create(TestActor.class));
+ TestActor testActor = (TestActor) testActorRef.underlyingActor();
+
+ final Counter c = new SimpleCounter();
+ final Gauge<String> g = new Gauge<String>() {
+ @Override
+ public String getValue() {
+ return "Hello";
+ }
+ };
+ final Histogram h = new TestHistogram();
+ final Meter m = new Meter() {
+
+ @Override
+ public void markEvent() {
+ }
+
+ @Override
+ public void markEvent(long n) {
+ }
+
+ @Override
+ public double getRate() {
+ return 5;
+ }
+
+ @Override
+ public long getCount() {
+ return 10;
+ }
+ };
+
+ MetricRegistryImpl registry = new MetricRegistryImpl(MetricRegistryConfiguration.defaultMetricRegistryConfiguration());
+ final TaskManagerMetricGroup tm = new TaskManagerMetricGroup(registry, "host", "id");
+
+ MetricQueryService.notifyOfAddedMetric(serviceActor, c, "counter", tm);
+ MetricQueryService.notifyOfAddedMetric(serviceActor, g, "gauge", tm);
+ MetricQueryService.notifyOfAddedMetric(serviceActor, h, "histogram", tm);
+ MetricQueryService.notifyOfAddedMetric(serviceActor, m, "meter", tm);
+ serviceActor.tell(MetricQueryService.getCreateDump(), testActorRef);
+ synchronized (testActor.lock) {
+ if (testActor.message == null) {
+ testActor.lock.wait();
+ }
}
- };
- final Histogram h = new TestHistogram();
- final Meter m = new Meter() {
- @Override
- public void markEvent() {
+ MetricDumpSerialization.MetricSerializationResult dump = (MetricDumpSerialization.MetricSerializationResult) testActor.message;
+ testActor.message = null;
+ assertTrue(dump.serializedCounters.length > 0);
+ assertTrue(dump.serializedGauges.length > 0);
+ assertTrue(dump.serializedHistograms.length > 0);
+ assertTrue(dump.serializedMeters.length > 0);
+
+ MetricQueryService.notifyOfRemovedMetric(serviceActor, c);
+ MetricQueryService.notifyOfRemovedMetric(serviceActor, g);
+ MetricQueryService.notifyOfRemovedMetric(serviceActor, h);
+ MetricQueryService.notifyOfRemovedMetric(serviceActor, m);
+
+ serviceActor.tell(MetricQueryService.getCreateDump(), testActorRef);
+ synchronized (testActor.lock) {
+ if (testActor.message == null) {
+ testActor.lock.wait();
+ }
}
- @Override
- public void markEvent(long n) {
- }
-
- @Override
- public double getRate() {
- return 5;
- }
-
- @Override
- public long getCount() {
- return 10;
- }
- };
-
- MetricRegistryImpl registry = new MetricRegistryImpl(MetricRegistryConfiguration.defaultMetricRegistryConfiguration());
- final TaskManagerMetricGroup tm = new TaskManagerMetricGroup(registry, "host", "id");
-
- MetricQueryService.notifyOfAddedMetric(serviceActor, c, "counter", tm);
- MetricQueryService.notifyOfAddedMetric(serviceActor, g, "gauge", tm);
- MetricQueryService.notifyOfAddedMetric(serviceActor, h, "histogram", tm);
- MetricQueryService.notifyOfAddedMetric(serviceActor, m, "meter", tm);
- serviceActor.tell(MetricQueryService.getCreateDump(), testActorRef);
- synchronized (testActor.lock) {
- if (testActor.message == null) {
- testActor.lock.wait();
- }
+ MetricDumpSerialization.MetricSerializationResult emptyDump = (MetricDumpSerialization.MetricSerializationResult) testActor.message;
+ testActor.message = null;
+ assertEquals(0, emptyDump.serializedCounters.length);
+ assertEquals(0, emptyDump.serializedGauges.length);
+ assertEquals(0, emptyDump.serializedHistograms.length);
+ assertEquals(0, emptyDump.serializedMeters.length);
+ } finally {
+ s.terminate();
}
-
- MetricDumpSerialization.MetricSerializationResult dump = (MetricDumpSerialization.MetricSerializationResult) testActor.message;
- testActor.message = null;
- assertTrue(dump.serializedCounters.length > 0);
- assertTrue(dump.serializedGauges.length > 0);
- assertTrue(dump.serializedHistograms.length > 0);
- assertTrue(dump.serializedMeters.length > 0);
-
- MetricQueryService.notifyOfRemovedMetric(serviceActor, c);
- MetricQueryService.notifyOfRemovedMetric(serviceActor, g);
- MetricQueryService.notifyOfRemovedMetric(serviceActor, h);
- MetricQueryService.notifyOfRemovedMetric(serviceActor, m);
-
- serviceActor.tell(MetricQueryService.getCreateDump(), testActorRef);
- synchronized (testActor.lock) {
- if (testActor.message == null) {
- testActor.lock.wait();
- }
- }
-
- MetricDumpSerialization.MetricSerializationResult emptyDump = (MetricDumpSerialization.MetricSerializationResult) testActor.message;
- testActor.message = null;
- assertEquals(0, emptyDump.serializedCounters.length);
- assertEquals(0, emptyDump.serializedGauges.length);
- assertEquals(0, emptyDump.serializedHistograms.length);
- assertEquals(0, emptyDump.serializedMeters.length);
-
- s.terminate();
}
@Test
public void testHandleOversizedMetricMessage() throws Exception {
ActorSystem s = AkkaUtils.createLocalActorSystem(new Configuration());
- final long sizeLimit = 200L;
- ActorRef serviceActor = MetricQueryService.startMetricQueryService(s, null, sizeLimit);
- TestActorRef testActorRef = TestActorRef.create(s, Props.create(TestActor.class));
- TestActor testActor = (TestActor) testActorRef.underlyingActor();
+ try {
+ final long sizeLimit = 200L;
+ ActorRef serviceActor = MetricQueryService.startMetricQueryService(s, null, sizeLimit);
+ TestActorRef testActorRef = TestActorRef.create(s, Props.create(TestActor.class));
+ TestActor testActor = (TestActor) testActorRef.underlyingActor();
+
+ final Counter c = new SimpleCounter();
+ final Histogram h = new TestHistogram();
+ final Meter m = new Meter() {
+
+ @Override
+ public void markEvent() {
+ }
+
+ @Override
+ public void markEvent(long n) {
+ }
+
+ @Override
+ public double getRate() {
+ return 5;
+ }
+
+ @Override
+ public long getCount() {
+ return 10;
+ }
+ };
+
+ MetricRegistryImpl registry = new MetricRegistryImpl(MetricRegistryConfiguration.defaultMetricRegistryConfiguration());
+ final TaskManagerMetricGroup tm = new TaskManagerMetricGroup(registry, "host", "id");
+
+ final String gaugeValue = "Hello";
+ final long requiredGaugesToExceedLimit = sizeLimit / gaugeValue.length() + 1;
+ List<Tuple2<String, Gauge<String>>> gauges = LongStream.range(0, requiredGaugesToExceedLimit)
+ .mapToObj(x -> Tuple2.of("gauge" + x, (Gauge<String>) () -> "Hello" + x))
+ .collect(Collectors.toList());
+ gauges.forEach(gauge -> MetricQueryService.notifyOfAddedMetric(serviceActor, gauge.f1, gauge.f0, tm));
+
+ MetricQueryService.notifyOfAddedMetric(serviceActor, c, "counter", tm);
+ MetricQueryService.notifyOfAddedMetric(serviceActor, h, "histogram", tm);
+ MetricQueryService.notifyOfAddedMetric(serviceActor, m, "meter", tm);
+
+ serviceActor.tell(MetricQueryService.getCreateDump(), testActorRef);
+ synchronized (testActor.lock) {
+ if (testActor.message == null) {
+ testActor.lock.wait();
+ }
+ }
- final Counter c = new SimpleCounter();
- final Histogram h = new TestHistogram();
- final Meter m = new Meter() {
+ MetricDumpSerialization.MetricSerializationResult dump = (MetricDumpSerialization.MetricSerializationResult) testActor.message;
+ testActor.message = null;
+ assertTrue(dump.serializedCounters.length > 0);
+ assertEquals(1, dump.numCounters);
+ assertTrue(dump.serializedMeters.length > 0);
+ assertEquals(1, dump.numMeters);
- @Override
- public void markEvent() {
- }
+ // gauges exceeded the size limit and will be excluded
+ assertEquals(0, dump.serializedGauges.length);
+ assertEquals(0, dump.numGauges);
- @Override
- public void markEvent(long n) {
- }
+ assertTrue(dump.serializedHistograms.length > 0);
+ assertEquals(1, dump.numHistograms);
- @Override
- public double getRate() {
- return 5;
+ // unregister all but one gauge to ensure gauges are reported again if the remaining fit
+ for (int x = 1; x < gauges.size(); x++) {
+ MetricQueryService.notifyOfRemovedMetric(serviceActor, gauges.get(x).f1);
}
- @Override
- public long getCount() {
- return 10;
- }
- };
-
- MetricRegistryImpl registry = new MetricRegistryImpl(MetricRegistryConfiguration.defaultMetricRegistryConfiguration());
- final TaskManagerMetricGroup tm = new TaskManagerMetricGroup(registry, "host", "id");
-
- final String gaugeValue = "Hello";
- final long requiredGaugesToExceedLimit = sizeLimit / gaugeValue.length() + 1;
- List<Tuple2<String, Gauge<String>>> gauges = LongStream.range(0, requiredGaugesToExceedLimit)
- .mapToObj(x -> Tuple2.of("gauge" + x, (Gauge<String>) () -> "Hello" + x))
- .collect(Collectors.toList());
- gauges.forEach(gauge -> MetricQueryService.notifyOfAddedMetric(serviceActor, gauge.f1, gauge.f0, tm));
-
- MetricQueryService.notifyOfAddedMetric(serviceActor, c, "counter", tm);
- MetricQueryService.notifyOfAddedMetric(serviceActor, h, "histogram", tm);
- MetricQueryService.notifyOfAddedMetric(serviceActor, m, "meter", tm);
-
- serviceActor.tell(MetricQueryService.getCreateDump(), testActorRef);
- synchronized (testActor.lock) {
- if (testActor.message == null) {
- testActor.lock.wait();
+ serviceActor.tell(MetricQueryService.getCreateDump(), testActorRef);
+ synchronized (testActor.lock) {
+ if (testActor.message == null) {
+ testActor.lock.wait();
+ }
}
- }
-
- MetricDumpSerialization.MetricSerializationResult dump = (MetricDumpSerialization.MetricSerializationResult) testActor.message;
- testActor.message = null;
- assertTrue(dump.serializedCounters.length > 0);
- assertEquals(1, dump.numCounters);
- assertTrue(dump.serializedMeters.length > 0);
- assertEquals(1, dump.numMeters);
-
- // gauges exceeded the size limit and will be excluded
- assertEquals(0, dump.serializedGauges.length);
- assertEquals(0, dump.numGauges);
-
- assertTrue(dump.serializedHistograms.length > 0);
- assertEquals(1, dump.numHistograms);
- // unregister all but one gauge to ensure gauges are reported again if the remaining fit
- for (int x = 1; x < gauges.size(); x++) {
- MetricQueryService.notifyOfRemovedMetric(serviceActor, gauges.get(x).f1);
+ MetricDumpSerialization.MetricSerializationResult recoveredDump = (MetricDumpSerialization.MetricSerializationResult) testActor.message;
+ testActor.message = null;
+
+ assertTrue(recoveredDump.serializedCounters.length > 0);
+ assertEquals(1, recoveredDump.numCounters);
+ assertTrue(recoveredDump.serializedMeters.length > 0);
+ assertEquals(1, recoveredDump.numMeters);
+ assertTrue(recoveredDump.serializedGauges.length > 0);
+ assertEquals(1, recoveredDump.numGauges);
+ assertTrue(recoveredDump.serializedHistograms.length > 0);
+ assertEquals(1, recoveredDump.numHistograms);
+ } finally {
+ s.terminate();
}
-
- serviceActor.tell(MetricQueryService.getCreateDump(), testActorRef);
- synchronized (testActor.lock) {
- if (testActor.message == null) {
- testActor.lock.wait();
- }
- }
-
- MetricDumpSerialization.MetricSerializationResult recoveredDump = (MetricDumpSerialization.MetricSerializationResult) testActor.message;
- testActor.message = null;
-
- assertTrue(recoveredDump.serializedCounters.length > 0);
- assertEquals(1, recoveredDump.numCounters);
- assertTrue(recoveredDump.serializedMeters.length > 0);
- assertEquals(1, recoveredDump.numMeters);
- assertTrue(recoveredDump.serializedGauges.length > 0);
- assertEquals(1, recoveredDump.numGauges);
- assertTrue(recoveredDump.serializedHistograms.length > 0);
- assertEquals(1, recoveredDump.numHistograms);
-
- s.terminate();
}
private static class TestActor extends UntypedActor {