@Test public void test_restart() throws InterruptedException { mockSuccessfulServerRestart(); RestartProgress restartProgress = restartService.restartNodes(cluster.getEndpoints(null), Duration.ofSeconds(2), STATES); assertThat(restartProgress.getErrors().size(), is(equalTo(0))); Map<Endpoint, LogicalServerState> restarted = restartProgress.await(Duration.ofSeconds(10)); assertThat(restarted.toString(), restarted.size(), is(equalTo(6))); IntStream.of(PORTS).forEach(port -> { verify(diagnosticServiceMock("localhost", port)).getProxy(DynamicConfigService.class); verify(dynamicConfigServiceMock("localhost", port)).restart(any()); verify(diagnosticServiceMock("localhost", port)).getLogicalServerState(); }); }
public RestartProgress restartNodes(Collection<Node.Endpoint> endpoints, Duration restartDelay, Collection<LogicalServerState> acceptedStates) { if (restartDelay.getSeconds() < 1) { throw new IllegalArgumentException("Restart delay must be at least 1 second"); } LOGGER.debug("Asking all nodes: {} to restart themselves", endpoints); Collection<Node.Endpoint> restartRequested = new HashSet<>(); Map<Node.Endpoint, Exception> restartRequestFailed = new HashMap<>(); for (Node.Endpoint endpoint : endpoints) { try (DiagnosticService diagnosticService = diagnosticServiceProvider.fetchDiagnosticService(endpoint.getAddress())) { diagnosticService.getProxy(DynamicConfigService.class).restart(restartDelay); restartRequested.add(endpoint); } catch (Exception e) { restartRequestFailed.put(endpoint, e); LOGGER.debug("Failed asking node {} to restart: {}", endpoint, e.getMessage(), e); } } CountDownLatch done = new CountDownLatch(restartRequested.size()); Map<Node.Endpoint, LogicalServerState> restartedNodes = new ConcurrentHashMap<>(); AtomicReference<BiConsumer<Node.Endpoint, LogicalServerState>> progressCallback = new AtomicReference<>(); AtomicBoolean continuePolling = new AtomicBoolean(true); ExecutorService executorService = Executors.newFixedThreadPool(concurrencySizing.getThreadCount(endpoints.size()), r -> new Thread(r, getClass().getName())); restartRequested.forEach(endpoint -> executorService.submit(() -> { try { Thread.sleep(restartDelay.toMillis() + 5_000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } LogicalServerState state = null; while (state == null && continuePolling.get() && !Thread.currentThread().isInterrupted()) { try { state = isRestarted(endpoint, acceptedStates); if (state != null) { LOGGER.debug("Node: {} has restarted", endpoint); restartedNodes.put(endpoint, state); BiConsumer<Node.Endpoint, LogicalServerState> cb = progressCallback.get(); if (cb != null) { cb.accept(endpoint, state); } done.countDown(); } else { Thread.sleep(1_000); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } })); return new RestartProgress() { @Override public void await() throws InterruptedException { try { done.await(); } finally { continuePolling.set(false); shutdown(executorService); } } @Override @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED") public Map<Node.Endpoint, LogicalServerState> await(Duration duration) throws InterruptedException { try { done.await(duration.toMillis(), MILLISECONDS); return new HashMap<>(restartedNodes); } finally { continuePolling.set(false); shutdown(executorService); } } @Override public void onRestarted(BiConsumer<Node.Endpoint, LogicalServerState> c) { progressCallback.set(c); restartedNodes.forEach(c); } @Override public Map<Node.Endpoint, Exception> getErrors() { return restartRequestFailed; } }; }
RestartService { public RestartProgress restartNodes(Collection<Node.Endpoint> endpoints, Duration restartDelay, Collection<LogicalServerState> acceptedStates) { if (restartDelay.getSeconds() < 1) { throw new IllegalArgumentException("Restart delay must be at least 1 second"); } LOGGER.debug("Asking all nodes: {} to restart themselves", endpoints); Collection<Node.Endpoint> restartRequested = new HashSet<>(); Map<Node.Endpoint, Exception> restartRequestFailed = new HashMap<>(); for (Node.Endpoint endpoint : endpoints) { try (DiagnosticService diagnosticService = diagnosticServiceProvider.fetchDiagnosticService(endpoint.getAddress())) { diagnosticService.getProxy(DynamicConfigService.class).restart(restartDelay); restartRequested.add(endpoint); } catch (Exception e) { restartRequestFailed.put(endpoint, e); LOGGER.debug("Failed asking node {} to restart: {}", endpoint, e.getMessage(), e); } } CountDownLatch done = new CountDownLatch(restartRequested.size()); Map<Node.Endpoint, LogicalServerState> restartedNodes = new ConcurrentHashMap<>(); AtomicReference<BiConsumer<Node.Endpoint, LogicalServerState>> progressCallback = new AtomicReference<>(); AtomicBoolean continuePolling = new AtomicBoolean(true); ExecutorService executorService = Executors.newFixedThreadPool(concurrencySizing.getThreadCount(endpoints.size()), r -> new Thread(r, getClass().getName())); restartRequested.forEach(endpoint -> executorService.submit(() -> { try { Thread.sleep(restartDelay.toMillis() + 5_000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } LogicalServerState state = null; while (state == null && continuePolling.get() && !Thread.currentThread().isInterrupted()) { try { state = isRestarted(endpoint, acceptedStates); if (state != null) { LOGGER.debug("Node: {} has restarted", endpoint); restartedNodes.put(endpoint, state); BiConsumer<Node.Endpoint, LogicalServerState> cb = progressCallback.get(); if (cb != null) { cb.accept(endpoint, state); } done.countDown(); } else { Thread.sleep(1_000); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } })); return new RestartProgress() { @Override public void await() throws InterruptedException { try { done.await(); } finally { continuePolling.set(false); shutdown(executorService); } } @Override @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED") public Map<Node.Endpoint, LogicalServerState> await(Duration duration) throws InterruptedException { try { done.await(duration.toMillis(), MILLISECONDS); return new HashMap<>(restartedNodes); } finally { continuePolling.set(false); shutdown(executorService); } } @Override public void onRestarted(BiConsumer<Node.Endpoint, LogicalServerState> c) { progressCallback.set(c); restartedNodes.forEach(c); } @Override public Map<Node.Endpoint, Exception> getErrors() { return restartRequestFailed; } }; } }
RestartService { public RestartProgress restartNodes(Collection<Node.Endpoint> endpoints, Duration restartDelay, Collection<LogicalServerState> acceptedStates) { if (restartDelay.getSeconds() < 1) { throw new IllegalArgumentException("Restart delay must be at least 1 second"); } LOGGER.debug("Asking all nodes: {} to restart themselves", endpoints); Collection<Node.Endpoint> restartRequested = new HashSet<>(); Map<Node.Endpoint, Exception> restartRequestFailed = new HashMap<>(); for (Node.Endpoint endpoint : endpoints) { try (DiagnosticService diagnosticService = diagnosticServiceProvider.fetchDiagnosticService(endpoint.getAddress())) { diagnosticService.getProxy(DynamicConfigService.class).restart(restartDelay); restartRequested.add(endpoint); } catch (Exception e) { restartRequestFailed.put(endpoint, e); LOGGER.debug("Failed asking node {} to restart: {}", endpoint, e.getMessage(), e); } } CountDownLatch done = new CountDownLatch(restartRequested.size()); Map<Node.Endpoint, LogicalServerState> restartedNodes = new ConcurrentHashMap<>(); AtomicReference<BiConsumer<Node.Endpoint, LogicalServerState>> progressCallback = new AtomicReference<>(); AtomicBoolean continuePolling = new AtomicBoolean(true); ExecutorService executorService = Executors.newFixedThreadPool(concurrencySizing.getThreadCount(endpoints.size()), r -> new Thread(r, getClass().getName())); restartRequested.forEach(endpoint -> executorService.submit(() -> { try { Thread.sleep(restartDelay.toMillis() + 5_000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } LogicalServerState state = null; while (state == null && continuePolling.get() && !Thread.currentThread().isInterrupted()) { try { state = isRestarted(endpoint, acceptedStates); if (state != null) { LOGGER.debug("Node: {} has restarted", endpoint); restartedNodes.put(endpoint, state); BiConsumer<Node.Endpoint, LogicalServerState> cb = progressCallback.get(); if (cb != null) { cb.accept(endpoint, state); } done.countDown(); } else { Thread.sleep(1_000); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } })); return new RestartProgress() { @Override public void await() throws InterruptedException { try { done.await(); } finally { continuePolling.set(false); shutdown(executorService); } } @Override @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED") public Map<Node.Endpoint, LogicalServerState> await(Duration duration) throws InterruptedException { try { done.await(duration.toMillis(), MILLISECONDS); return new HashMap<>(restartedNodes); } finally { continuePolling.set(false); shutdown(executorService); } } @Override public void onRestarted(BiConsumer<Node.Endpoint, LogicalServerState> c) { progressCallback.set(c); restartedNodes.forEach(c); } @Override public Map<Node.Endpoint, Exception> getErrors() { return restartRequestFailed; } }; } RestartService(DiagnosticServiceProvider diagnosticServiceProvider, ConcurrencySizing concurrencySizing); }
RestartService { public RestartProgress restartNodes(Collection<Node.Endpoint> endpoints, Duration restartDelay, Collection<LogicalServerState> acceptedStates) { if (restartDelay.getSeconds() < 1) { throw new IllegalArgumentException("Restart delay must be at least 1 second"); } LOGGER.debug("Asking all nodes: {} to restart themselves", endpoints); Collection<Node.Endpoint> restartRequested = new HashSet<>(); Map<Node.Endpoint, Exception> restartRequestFailed = new HashMap<>(); for (Node.Endpoint endpoint : endpoints) { try (DiagnosticService diagnosticService = diagnosticServiceProvider.fetchDiagnosticService(endpoint.getAddress())) { diagnosticService.getProxy(DynamicConfigService.class).restart(restartDelay); restartRequested.add(endpoint); } catch (Exception e) { restartRequestFailed.put(endpoint, e); LOGGER.debug("Failed asking node {} to restart: {}", endpoint, e.getMessage(), e); } } CountDownLatch done = new CountDownLatch(restartRequested.size()); Map<Node.Endpoint, LogicalServerState> restartedNodes = new ConcurrentHashMap<>(); AtomicReference<BiConsumer<Node.Endpoint, LogicalServerState>> progressCallback = new AtomicReference<>(); AtomicBoolean continuePolling = new AtomicBoolean(true); ExecutorService executorService = Executors.newFixedThreadPool(concurrencySizing.getThreadCount(endpoints.size()), r -> new Thread(r, getClass().getName())); restartRequested.forEach(endpoint -> executorService.submit(() -> { try { Thread.sleep(restartDelay.toMillis() + 5_000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } LogicalServerState state = null; while (state == null && continuePolling.get() && !Thread.currentThread().isInterrupted()) { try { state = isRestarted(endpoint, acceptedStates); if (state != null) { LOGGER.debug("Node: {} has restarted", endpoint); restartedNodes.put(endpoint, state); BiConsumer<Node.Endpoint, LogicalServerState> cb = progressCallback.get(); if (cb != null) { cb.accept(endpoint, state); } done.countDown(); } else { Thread.sleep(1_000); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } })); return new RestartProgress() { @Override public void await() throws InterruptedException { try { done.await(); } finally { continuePolling.set(false); shutdown(executorService); } } @Override @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED") public Map<Node.Endpoint, LogicalServerState> await(Duration duration) throws InterruptedException { try { done.await(duration.toMillis(), MILLISECONDS); return new HashMap<>(restartedNodes); } finally { continuePolling.set(false); shutdown(executorService); } } @Override public void onRestarted(BiConsumer<Node.Endpoint, LogicalServerState> c) { progressCallback.set(c); restartedNodes.forEach(c); } @Override public Map<Node.Endpoint, Exception> getErrors() { return restartRequestFailed; } }; } RestartService(DiagnosticServiceProvider diagnosticServiceProvider, ConcurrencySizing concurrencySizing); RestartProgress restartNodes(Collection<Node.Endpoint> endpoints, Duration restartDelay, Collection<LogicalServerState> acceptedStates); }
RestartService { public RestartProgress restartNodes(Collection<Node.Endpoint> endpoints, Duration restartDelay, Collection<LogicalServerState> acceptedStates) { if (restartDelay.getSeconds() < 1) { throw new IllegalArgumentException("Restart delay must be at least 1 second"); } LOGGER.debug("Asking all nodes: {} to restart themselves", endpoints); Collection<Node.Endpoint> restartRequested = new HashSet<>(); Map<Node.Endpoint, Exception> restartRequestFailed = new HashMap<>(); for (Node.Endpoint endpoint : endpoints) { try (DiagnosticService diagnosticService = diagnosticServiceProvider.fetchDiagnosticService(endpoint.getAddress())) { diagnosticService.getProxy(DynamicConfigService.class).restart(restartDelay); restartRequested.add(endpoint); } catch (Exception e) { restartRequestFailed.put(endpoint, e); LOGGER.debug("Failed asking node {} to restart: {}", endpoint, e.getMessage(), e); } } CountDownLatch done = new CountDownLatch(restartRequested.size()); Map<Node.Endpoint, LogicalServerState> restartedNodes = new ConcurrentHashMap<>(); AtomicReference<BiConsumer<Node.Endpoint, LogicalServerState>> progressCallback = new AtomicReference<>(); AtomicBoolean continuePolling = new AtomicBoolean(true); ExecutorService executorService = Executors.newFixedThreadPool(concurrencySizing.getThreadCount(endpoints.size()), r -> new Thread(r, getClass().getName())); restartRequested.forEach(endpoint -> executorService.submit(() -> { try { Thread.sleep(restartDelay.toMillis() + 5_000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } LogicalServerState state = null; while (state == null && continuePolling.get() && !Thread.currentThread().isInterrupted()) { try { state = isRestarted(endpoint, acceptedStates); if (state != null) { LOGGER.debug("Node: {} has restarted", endpoint); restartedNodes.put(endpoint, state); BiConsumer<Node.Endpoint, LogicalServerState> cb = progressCallback.get(); if (cb != null) { cb.accept(endpoint, state); } done.countDown(); } else { Thread.sleep(1_000); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } })); return new RestartProgress() { @Override public void await() throws InterruptedException { try { done.await(); } finally { continuePolling.set(false); shutdown(executorService); } } @Override @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED") public Map<Node.Endpoint, LogicalServerState> await(Duration duration) throws InterruptedException { try { done.await(duration.toMillis(), MILLISECONDS); return new HashMap<>(restartedNodes); } finally { continuePolling.set(false); shutdown(executorService); } } @Override public void onRestarted(BiConsumer<Node.Endpoint, LogicalServerState> c) { progressCallback.set(c); restartedNodes.forEach(c); } @Override public Map<Node.Endpoint, Exception> getErrors() { return restartRequestFailed; } }; } RestartService(DiagnosticServiceProvider diagnosticServiceProvider, ConcurrencySizing concurrencySizing); RestartProgress restartNodes(Collection<Node.Endpoint> endpoints, Duration restartDelay, Collection<LogicalServerState> acceptedStates);  }