class ConnectionTarget {
        private final DiscoveryNode discoveryNode;

        private CompletableFuture<Void> future = new CompletableFuture<>();
        private ActivityType activityType = ActivityType.IDLE; // indicates what any listeners are awaiting

        private final AtomicInteger consecutiveFailureCount = new AtomicInteger();

        private final Runnable connectActivity = new AbstractRunnable() {

            final AbstractRunnable abstractRunnable = this;

            @Override
            protected void doRun() {
                assert Thread.holdsLock(mutex) == false : "mutex unexpectedly held";
                if (transportService.nodeConnected(discoveryNode)) {
                    // transportService.connectToNode is a no-op if already connected, but we don't want any DEBUG logging in this case
                    // since we run this for every node on every cluster state update.
                    LOGGER.trace("still connected to {}", discoveryNode);
                    onConnected();
                } else {
                    LOGGER.debug("connecting to {}", discoveryNode);
                    transportService.connectToNode(discoveryNode, new ActionListener<Void>() {
                        @Override
                        public void onResponse(Void aVoid) {
                            assert Thread.holdsLock(mutex) == false : "mutex unexpectedly held";
                            LOGGER.debug("connected to {}", discoveryNode);
                            onConnected();
                        }

                        @Override
                        public void onFailure(Exception e) {
                            abstractRunnable.onFailure(e);
                        }
                    });
                }
            }

            private void onConnected() {
                consecutiveFailureCount.set(0);
                onCompletion(ActivityType.CONNECTING, null, disconnectActivity);
            }

            @Override
            public void onFailure(Exception e) {
                assert Thread.holdsLock(mutex) == false : "mutex unexpectedly held";
                final int currentFailureCount = consecutiveFailureCount.incrementAndGet();
                if (!lifecycle.stoppedOrClosed()) {
                    // only warn every 6th failure
                    final Level level = currentFailureCount % 6 == 0 ? Level.WARN : Level.DEBUG;
                    LOGGER.log(level, new ParameterizedMessage("failed to connect to {} (tried [{}] times lifecycle={})",
                        discoveryNode, currentFailureCount, lifecycle), e);
                }
                onCompletion(ActivityType.CONNECTING, e, disconnectActivity);
            }

            @Override
            public String toString() {
                return "connect to " + discoveryNode;
            }
        };

        private final Runnable disconnectActivity = new AbstractRunnable() {
            @Override
            protected void doRun() {
                assert Thread.holdsLock(mutex) == false : "mutex unexpectedly held";
                transportService.disconnectFromNode(discoveryNode);
                consecutiveFailureCount.set(0);
                LOGGER.debug("disconnected from {}", discoveryNode);
                onCompletion(ActivityType.DISCONNECTING, null, connectActivity);
            }

            @Override
            public void onFailure(Exception e) {
                assert Thread.holdsLock(mutex) == false : "mutex unexpectedly held";
                consecutiveFailureCount.incrementAndGet();
                // we may not have disconnected, but will not retry, so this connection might have leaked
                LOGGER.warn(new ParameterizedMessage("failed to disconnect from {}, possible connection leak", discoveryNode), e);
                assert false : "failed to disconnect from " + discoveryNode + ", possible connection leak\n" + e;
                onCompletion(ActivityType.DISCONNECTING, e, connectActivity);
            }
        };

        ConnectionTarget(DiscoveryNode discoveryNode) {
            this.discoveryNode = discoveryNode;
        }

        Runnable connect(@Nullable ActionListener<Void> listener) {
            return addListenerAndStartActivity(listener, ActivityType.CONNECTING, connectActivity,
                "disconnection cancelled by reconnection");
        }

        Runnable disconnect() {
            return addListenerAndStartActivity(null, ActivityType.DISCONNECTING, disconnectActivity,
                "connection cancelled by disconnection");
        }

        Runnable ensureConnected(@Nullable ActionListener<Void> listener) {
            assert Thread.holdsLock(mutex) : "mutex not held";

            if (activityType == ActivityType.IDLE) {
                if (transportService.nodeConnected(discoveryNode)) {
                    return () -> listener.onResponse(null);
                } else {
                    // target is disconnected, and we are currently idle, so start a connection process.
                    activityType = ActivityType.CONNECTING;
                    addListener(listener);
                    return connectActivity;
                }
            } else {
                addListener(listener);
                return () -> {
                };
            }
        }

        Runnable awaitCurrentActivity(ActionListener<Void> listener) {
            assert Thread.holdsLock(mutex) : "mutex not held";

            if (activityType == ActivityType.IDLE) {
                return () -> listener.onResponse(null);
            } else {
                addListener(listener);
                return () -> {
                };
            }
        }

        private void addListener(@Nullable ActionListener<Void> listener) {
            assert Thread.holdsLock(mutex) : "mutex not held";
            assert activityType != ActivityType.IDLE;
            if (listener != null) {
                future.whenComplete(listener);
            }
        }

        private CompletableFuture<Void> getAndClearFuture() {
            assert Thread.holdsLock(mutex) : "mutex not held";
            final CompletableFuture<Void> drainedFuture = future;
            future = new CompletableFuture<>();
            return drainedFuture;
        }

        private Runnable addListenerAndStartActivity(@Nullable ActionListener<Void> listener, ActivityType newActivityType,
                                                     Runnable activity, String cancellationMessage) {<FILL_FUNCTION_BODY>}

        private void onCompletion(ActivityType completedActivityType, @Nullable Exception e, Runnable oppositeActivity) {
            assert Thread.holdsLock(mutex) == false : "mutex unexpectedly held";

            final Runnable cleanup;
            synchronized (mutex) {
                assert activityType != ActivityType.IDLE;
                if (activityType == completedActivityType) {
                    final CompletableFuture<Void> oldFuture = getAndClearFuture();
                    activityType = ActivityType.IDLE;

                    cleanup = e == null ? () -> oldFuture.complete(null) : () -> oldFuture.completeExceptionally(e);

                    if (completedActivityType.equals(ActivityType.DISCONNECTING)) {
                        final ConnectionTarget removedTarget = targetsByNode.remove(discoveryNode);
                        assert removedTarget == this : removedTarget + " vs " + this;
                    }
                } else {
                    cleanup = oppositeActivity;
                }
            }
            cleanup.run();
        }

        boolean isPendingDisconnection() {
            assert Thread.holdsLock(mutex) : "mutex not held";
            return activityType == ActivityType.DISCONNECTING;
        }

        @Override
        public String toString() {
            synchronized (mutex) {
                return "ConnectionTarget{" +
                    "discoveryNode=" + discoveryNode +
                    ", activityType=" + activityType +
                    '}';
            }
        }
    }

            assert Thread.holdsLock(mutex) : "mutex not held";
            assert newActivityType.equals(ActivityType.IDLE) == false;

            if (activityType == ActivityType.IDLE) {
                activityType = newActivityType;
                addListener(listener);
                return activity;
            }

            if (activityType == newActivityType) {
                addListener(listener);
                return () -> {
                };
            }

            activityType = newActivityType;
            final CompletableFuture<Void> oldFuture = getAndClearFuture();
            addListener(listener);
            return () -> oldFuture.completeExceptionally(new ElasticsearchException(cancellationMessage));
        
<methods>public void addLifecycleListener(org.elasticsearch.common.component.LifecycleListener) ,public void close() ,public org.elasticsearch.common.component.Lifecycle.State lifecycleState() ,public void removeLifecycleListener(org.elasticsearch.common.component.LifecycleListener) ,public void start() ,public void stop() <variables>protected final org.elasticsearch.common.component.Lifecycle lifecycle,private final List<org.elasticsearch.common.component.LifecycleListener> listeners