void fenceZombieSourceTasks(final ConnectorTaskId id, Callback<Void> callback) {
        log.trace("Performing preflight zombie check for task {}", id);
        fenceZombieSourceTasks(id.connector(), (error, ignored) -> {
            if (error == null) {
                callback.onCompletion(null, null);
            } else if (error instanceof NotLeaderException) {
                if (restClient != null) {
                    String workerUrl = ((NotLeaderException) error).forwardUrl();
                    String fenceUrl = namespacedUrl(workerUrl)
                            .path("connectors")
                            .path(id.connector())
                            .path("fence")
                            .build()
                            .toString();
                    log.trace("Forwarding zombie fencing request for connector {} to leader at {}", id.connector(), fenceUrl);
                    forwardRequestExecutor.execute(() -> {
                        try {
                            String stageDescription = "Forwarding zombie fencing request to the leader at " + workerUrl;
                            try (TemporaryStage stage = new TemporaryStage(stageDescription, callback, time)) {
                                restClient.httpRequest(fenceUrl, "PUT", null, null, sessionKey, requestSignatureAlgorithm);
                            }
                            callback.onCompletion(null, null);
                        } catch (Throwable t) {
                            callback.onCompletion(t, null);
                        }
                    });
                } else {
                    callback.onCompletion(
                            new ConnectException(
                                    "This worker is not able to communicate with the leader of the cluster, "
                                            + "which is required for exactly-once source tasks. If running MirrorMaker 2 "
                                            + "in dedicated mode, consider enabling inter-worker communication via the "
                                            + "'dedicated.mode.enable.internal.rest' property."
                            ),
                            null
                    );
                }
            } else {
                error = ConnectUtils.maybeWrap(error, "Failed to perform zombie fencing");
                callback.onCompletion(error, null);
            }
        });
    }
Close resources managed by this herder but which are not explicitly started.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java