@Test public void testReporterScheduling() throws Exception { MetricConfig config = new MetricConfig(); config.setProperty("arg1", "hello"); config.setProperty(ConfigConstants.METRICS_REPORTER_INTERVAL_SUFFIX, "50 MILLISECONDS"); MetricRegistryImpl registry = new MetricRegistryImpl( MetricRegistryConfiguration.defaultMetricRegistryConfiguration(), Collections.singletonList(ReporterSetup.forReporter("test", config, new TestReporter3()))); long start = System.currentTimeMillis(); TestReporter3.reportCount = 0; for (int x = 0; x < 10; x++) { Thread.sleep(100); int reportCount = TestReporter3.reportCount; long curT = System.currentTimeMillis(); long maxAllowedReports = (curT - start) / 50 + 2; Assert.assertTrue("Too many reports were triggered.", maxAllowedReports >= reportCount); } Assert.assertTrue("No report was triggered.", TestReporter3.reportCount > 0); 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);  }