@Test public void testReporterNotifications() throws Exception { Configuration config = new Configuration(); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "test1." + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, TestReporter6.class.getName()); config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "test2." + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, TestReporter7.class.getName()); MetricRegistryImpl registry = new MetricRegistryImpl( MetricRegistryConfiguration.defaultMetricRegistryConfiguration(), Arrays.asList( ReporterSetup.forReporter("test1", new TestReporter6()), ReporterSetup.forReporter("test2", new TestReporter7()))); TaskManagerMetricGroup root = new TaskManagerMetricGroup(registry, "host", "id"); root.counter("rootCounter"); assertTrue(TestReporter6.addedMetric instanceof Counter); assertEquals("rootCounter", TestReporter6.addedMetricName); assertTrue(TestReporter7.addedMetric instanceof Counter); assertEquals("rootCounter", TestReporter7.addedMetricName); root.close(); assertTrue(TestReporter6.removedMetric instanceof Counter); assertEquals("rootCounter", TestReporter6.removedMetricName); assertTrue(TestReporter7.removedMetric instanceof Counter); assertEquals("rootCounter", TestReporter7.removedMetricName); registry.shutdown().get(); }
public CompletableFuture<Void> shutdown() { synchronized (lock) { if (isShutdown) { return terminationFuture; } else { isShutdown = true; final Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(3); final Time gracePeriod = Time.seconds(1L); if (metricQueryServiceRpcService != null) { final CompletableFuture<Void> metricQueryServiceRpcServiceTerminationFuture = metricQueryServiceRpcService.stopService(); terminationFutures.add(metricQueryServiceRpcServiceTerminationFuture); } Throwable throwable = null; for (MetricReporter reporter : reporters) { try { reporter.close(); } catch (Throwable t) { throwable = ExceptionUtils.firstOrSuppressed(t, throwable); } } reporters.clear(); if (throwable != null) { terminationFutures.add( FutureUtils.completedExceptionally( new FlinkException("Could not shut down the metric reporters properly.", throwable))); } final CompletableFuture<Void> executorShutdownFuture = ExecutorUtils.nonBlockingShutdown( gracePeriod.toMilliseconds(), TimeUnit.MILLISECONDS, executor); terminationFutures.add(executorShutdownFuture); FutureUtils .completeAll(terminationFutures) .whenComplete( (Void ignored, Throwable error) -> { if (error != null) { terminationFuture.completeExceptionally(error); } else { terminationFuture.complete(null); } }); return terminationFuture; } } }
MetricRegistryImpl implements MetricRegistry { public CompletableFuture<Void> shutdown() { synchronized (lock) { if (isShutdown) { return terminationFuture; } else { isShutdown = true; final Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(3); final Time gracePeriod = Time.seconds(1L); if (metricQueryServiceRpcService != null) { final CompletableFuture<Void> metricQueryServiceRpcServiceTerminationFuture = metricQueryServiceRpcService.stopService(); terminationFutures.add(metricQueryServiceRpcServiceTerminationFuture); } Throwable throwable = null; for (MetricReporter reporter : reporters) { try { reporter.close(); } catch (Throwable t) { throwable = ExceptionUtils.firstOrSuppressed(t, throwable); } } reporters.clear(); if (throwable != null) { terminationFutures.add( FutureUtils.completedExceptionally( new FlinkException("Could not shut down the metric reporters properly.", throwable))); } final CompletableFuture<Void> executorShutdownFuture = ExecutorUtils.nonBlockingShutdown( gracePeriod.toMilliseconds(), TimeUnit.MILLISECONDS, executor); terminationFutures.add(executorShutdownFuture); FutureUtils .completeAll(terminationFutures) .whenComplete( (Void ignored, Throwable error) -> { if (error != null) { terminationFuture.completeExceptionally(error); } else { terminationFuture.complete(null); } }); return terminationFuture; } } } }
MetricRegistryImpl implements MetricRegistry { public CompletableFuture<Void> shutdown() { synchronized (lock) { if (isShutdown) { return terminationFuture; } else { isShutdown = true; final Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(3); final Time gracePeriod = Time.seconds(1L); if (metricQueryServiceRpcService != null) { final CompletableFuture<Void> metricQueryServiceRpcServiceTerminationFuture = metricQueryServiceRpcService.stopService(); terminationFutures.add(metricQueryServiceRpcServiceTerminationFuture); } Throwable throwable = null; for (MetricReporter reporter : reporters) { try { reporter.close(); } catch (Throwable t) { throwable = ExceptionUtils.firstOrSuppressed(t, throwable); } } reporters.clear(); if (throwable != null) { terminationFutures.add( FutureUtils.completedExceptionally( new FlinkException("Could not shut down the metric reporters properly.", throwable))); } final CompletableFuture<Void> executorShutdownFuture = ExecutorUtils.nonBlockingShutdown( gracePeriod.toMilliseconds(), TimeUnit.MILLISECONDS, executor); terminationFutures.add(executorShutdownFuture); FutureUtils .completeAll(terminationFutures) .whenComplete( (Void ignored, Throwable error) -> { if (error != null) { terminationFuture.completeExceptionally(error); } else { terminationFuture.complete(null); } }); return terminationFuture; } } } MetricRegistryImpl(MetricRegistryConfiguration config);  MetricRegistryImpl(MetricRegistryConfiguration config, Collection<ReporterSetup> reporterConfigurations); }
MetricRegistryImpl implements MetricRegistry { public CompletableFuture<Void> shutdown() { synchronized (lock) { if (isShutdown) { return terminationFuture; } else { isShutdown = true; final Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(3); final Time gracePeriod = Time.seconds(1L); if (metricQueryServiceRpcService != null) { final CompletableFuture<Void> metricQueryServiceRpcServiceTerminationFuture = metricQueryServiceRpcService.stopService(); terminationFutures.add(metricQueryServiceRpcServiceTerminationFuture); } Throwable throwable = null; for (MetricReporter reporter : reporters) { try { reporter.close(); } catch (Throwable t) { throwable = ExceptionUtils.firstOrSuppressed(t, throwable); } } reporters.clear(); if (throwable != null) { terminationFutures.add( FutureUtils.completedExceptionally( new FlinkException("Could not shut down the metric reporters properly.", throwable))); } final CompletableFuture<Void> executorShutdownFuture = ExecutorUtils.nonBlockingShutdown( gracePeriod.toMilliseconds(), TimeUnit.MILLISECONDS, executor); terminationFutures.add(executorShutdownFuture); FutureUtils .completeAll(terminationFutures) .whenComplete( (Void ignored, Throwable error) -> { if (error != null) { terminationFuture.completeExceptionally(error); } else { terminationFuture.complete(null); } }); return terminationFuture; } } } MetricRegistryImpl(MetricRegistryConfiguration config);  MetricRegistryImpl(MetricRegistryConfiguration config, Collection<ReporterSetup> reporterConfigurations); void startQueryService(RpcService rpcService, ResourceID resourceID); @Nullable RpcService getMetricQueryServiceRpcService(); @Override @Nullable String getMetricQueryServiceGatewayRpcAddress(); @Override char getDelimiter(); @Override char getDelimiter(int reporterIndex); @Override int getNumberReporters(); @VisibleForTesting List<MetricReporter> getReporters(); boolean isShutdown(); CompletableFuture<Void> shutdown(); @Override ScopeFormats getScopeFormats(); @Override void register(Metric metric, String metricName, AbstractMetricGroup group); @Override void unregister(Metric metric, String metricName, AbstractMetricGroup group); }
MetricRegistryImpl implements MetricRegistry { public CompletableFuture<Void> shutdown() { synchronized (lock) { if (isShutdown) { return terminationFuture; } else { isShutdown = true; final Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(3); final Time gracePeriod = Time.seconds(1L); if (metricQueryServiceRpcService != null) { final CompletableFuture<Void> metricQueryServiceRpcServiceTerminationFuture = metricQueryServiceRpcService.stopService(); terminationFutures.add(metricQueryServiceRpcServiceTerminationFuture); } Throwable throwable = null; for (MetricReporter reporter : reporters) { try { reporter.close(); } catch (Throwable t) { throwable = ExceptionUtils.firstOrSuppressed(t, throwable); } } reporters.clear(); if (throwable != null) { terminationFutures.add( FutureUtils.completedExceptionally( new FlinkException("Could not shut down the metric reporters properly.", throwable))); } final CompletableFuture<Void> executorShutdownFuture = ExecutorUtils.nonBlockingShutdown( gracePeriod.toMilliseconds(), TimeUnit.MILLISECONDS, executor); terminationFutures.add(executorShutdownFuture); FutureUtils .completeAll(terminationFutures) .whenComplete( (Void ignored, Throwable error) -> { if (error != null) { terminationFuture.completeExceptionally(error); } else { terminationFuture.complete(null); } }); return terminationFuture; } } } MetricRegistryImpl(MetricRegistryConfiguration config);  MetricRegistryImpl(MetricRegistryConfiguration config, Collection<ReporterSetup> reporterConfigurations); void startQueryService(RpcService rpcService, ResourceID resourceID); @Nullable RpcService getMetricQueryServiceRpcService(); @Override @Nullable String getMetricQueryServiceGatewayRpcAddress(); @Override char getDelimiter(); @Override char getDelimiter(int reporterIndex); @Override int getNumberReporters(); @VisibleForTesting List<MetricReporter> getReporters(); boolean isShutdown(); CompletableFuture<Void> shutdown(); @Override ScopeFormats getScopeFormats(); @Override void register(Metric metric, String metricName, AbstractMetricGroup group); @Override void unregister(Metric metric, String metricName, AbstractMetricGroup group);  }