LeaderElector { protected static boolean loop(Consumer<CountDownLatch> consumer, long periodInMillis) { final ScheduledExecutorService singleThreadExecutorService = Executors.newSingleThreadScheduledExecutor(); final CountDownLatch countDownLatch = new CountDownLatch(1); final Future<?> future = singleThreadExecutorService.scheduleAtFixedRate( () -> consumer.accept(countDownLatch), 0L, periodInMillis, TimeUnit.MILLISECONDS); try { countDownLatch.await(); return true; } catch (InterruptedException e) { LOGGER.debug("Loop thread interrupted", e); Thread.currentThread().interrupt(); return false; } finally { future.cancel(true); singleThreadExecutorService.shutdownNow(); } } LeaderElector(C kubernetesClient, LeaderElectionConfig leaderElectionConfig); void run();  }
@Test void loopCompletesOkShouldShutdownExecutorService() throws Exception { final CountDownLatch signal = new CountDownLatch(1); final ExecutorService executor = Executors.newSingleThreadExecutor(); executor.submit(() -> loop(countDownLatch -> { countDownLatch.countDown(); signal.countDown(); }, 1)); signal.await(50, TimeUnit.MILLISECONDS); assertEquals(0, signal.getCount()); executor.shutdownNow(); }
@Test void loopInterruptedShouldShutdownExecutorService() throws Exception { final CountDownLatch signal = new CountDownLatch(1); final ExecutorService executor = Executors.newSingleThreadExecutor(); final AtomicReference<CountDownLatch> loopCountDownLatch = new AtomicReference<>(null); final Future<Boolean> futureLoop = executor.submit(() -> loop(countDownLatch -> { loopCountDownLatch.set(countDownLatch); signal.countDown(); }, 1)); signal.await(50, TimeUnit.MILLISECONDS); futureLoop.cancel(true); assertEquals(0, signal.getCount()); assertEquals(1, loopCountDownLatch.get().getCount()); executor.shutdownNow(); executor.awaitTermination(10, TimeUnit.SECONDS); }