class 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()) {
					// only log exceptions on debug or trace level
					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);
				}
			}

			// we couldn't resolve the gateway --> let's try again
			final CompletableFuture<T> newGatewayFuture = createGateway(getLeaderFuture());

			// let's check if there was a concurrent createNewFuture call
			if (atomicGatewayFuture.compareAndSet(currentGatewayFuture, newGatewayFuture)) {
				return newGatewayFuture;
			} else {
				return atomicGatewayFuture.get();
			}
		} else {
			return atomicGatewayFuture.get();
		}
	}

	public  LeaderGatewayRetriever();

}