LeaderGatewayRetriever extends LeaderRetriever implements GatewayRetriever<T> { @Override public CompletableFuture<T> getFuture() { final CompletableFuture<T> currentGatewayFuture = atomicGatewayFuture.get(); if (currentGatewayFuture.isCompletedExceptionally()) { try { currentGatewayFuture.get(); } catch (ExecutionException | InterruptedException executionException) { String leaderAddress; try { Tuple2<String, UUID> leaderAddressSessionId = getLeaderNow() .orElse(Tuple2.of("unknown address", HighAvailabilityServices.DEFAULT_LEADER_ID)); leaderAddress = leaderAddressSessionId.f0; } catch (Exception e) { log.warn("Could not obtain the current leader.", e); leaderAddress = "unknown leader address"; } if (log.isDebugEnabled() || log.isTraceEnabled()) { log.warn( "Error while retrieving the leader gateway. Retrying to connect to {}.", leaderAddress, ExceptionUtils.stripExecutionException(executionException)); } else { log.warn( "Error while retrieving the leader gateway. Retrying to connect to {}.", leaderAddress); } } final CompletableFuture<T> newGatewayFuture = createGateway(getLeaderFuture()); if (atomicGatewayFuture.compareAndSet(currentGatewayFuture, newGatewayFuture)) { return newGatewayFuture; } else { return atomicGatewayFuture.get(); } } else { return atomicGatewayFuture.get(); } } LeaderGatewayRetriever(); @Override CompletableFuture<T> getFuture(); @Override void notifyNewLeaderAddress(CompletableFuture<Tuple2<String, UUID>> newLeaderAddressFuture);  }
@Test public void testGatewayRetrievalFailures() throws Exception { final String address = "localhost"; final UUID leaderId = UUID.randomUUID(); RpcGateway rpcGateway = mock(RpcGateway.class); TestingLeaderGatewayRetriever leaderGatewayRetriever = new TestingLeaderGatewayRetriever(rpcGateway); TestingLeaderRetrievalService testingLeaderRetrievalService = new TestingLeaderRetrievalService(); testingLeaderRetrievalService.start(leaderGatewayRetriever); CompletableFuture<RpcGateway> gatewayFuture = leaderGatewayRetriever.getFuture(); testingLeaderRetrievalService.notifyListener(address, leaderId); try { gatewayFuture.get(); fail("The first future should have been failed."); } catch (ExecutionException ignored) { } assertFalse((leaderGatewayRetriever.getNow().isPresent())); assertEquals(rpcGateway, leaderGatewayRetriever.getNow().get()); }