GraphiteReporter implements MetricsReporter, Runnable { @Override public void close() { if (config.getBoolean(REPORTER_ENABLED)) { executor.submit(this); } executor.shutdown(); try { if (executor.awaitTermination(20, TimeUnit.SECONDS)) { log.debug("Executor was shut down successfully."); } else { log.error("Timed out before executor was shut down! It's possible some metrics data were not sent out!"); } } catch (InterruptedException e) { log.error("Unable to shutdown executor gracefully", e); } }  @Override void configure(final Map<String, ?> configs); @Override void init(List<KafkaMetric> metrics); @Override void metricChange(final KafkaMetric metric); @Override void metricRemoval(KafkaMetric metric); @Override void close(); @Override void run();  }
@Test public void testClose() throws InterruptedException { Map<String, Object> configs = initializeConfigWithReporter(); graphiteReporter.configure(configs); ScheduledExecutorService mockExecutor = mock(ScheduledExecutorService.class); Future mockFuture = mock(Future.class); when(mockExecutor.submit(graphiteReporter)).thenReturn(mockFuture); when(mockExecutor.awaitTermination(20, TimeUnit.SECONDS)).thenReturn(true); Whitebox.setInternalState(graphiteReporter, "executor", mockExecutor); graphiteReporter.close(); verify(mockExecutor).shutdown(); verify(mockExecutor).submit(graphiteReporter); verify(mockExecutor).awaitTermination(20, TimeUnit.SECONDS); }