private void doFenceZombieSourceTasks(String connName, Callback<Void> callback) {
        log.trace("Performing zombie fencing request for connector {}", connName);

        // We don't have to check for a pending rebalance here

        if (!isLeader())
            callback.onCompletion(new NotLeaderException("Only the leader may perform zombie fencing.", leaderUrl()), null);
        else if (!configState.contains(connName))
            callback.onCompletion(new NotFoundException("Connector " + connName + " not found"), null);
        else if (!isSourceConnector(connName))
            callback.onCompletion(new BadRequestException("Connector " + connName + " is not a source connector"), null);
        else {
            if (!refreshConfigSnapshot(workerSyncTimeoutMs)) {
                throw new ConnectException("Failed to read to end of config topic before performing zombie fencing");
            }

            int taskCount = configState.taskCount(connName);
            Integer taskCountRecord = configState.taskCountRecord(connName);

            ZombieFencing zombieFencing = null;
            boolean newFencing = false;
            synchronized (DistributedHerder.this) {
                // Check first to see if we have to do a fencing. The control flow is a little awkward here (why not stick this in
                // an else block lower down?) but we can't synchronize around the body below since that may contain a synchronous
                // write to the config topic.
                if (configState.pendingFencing(connName) && taskCountRecord != null
                        && (taskCountRecord != 1 || taskCount != 1)) {
                    int taskGen = configState.taskConfigGeneration(connName);
                    zombieFencing = activeZombieFencings.get(connName);
                    if (zombieFencing == null) {
                        zombieFencing = new ZombieFencing(connName, taskCountRecord, taskCount, taskGen);
                        activeZombieFencings.put(connName, zombieFencing);
                        newFencing = true;
                    }
                }
            }
            if (zombieFencing != null) {
                if (newFencing) {
                    zombieFencing.start();
                }
                zombieFencing.addCallback(callback);
                return;
            }

            if (!configState.pendingFencing(connName)) {
                // If the latest task count record for the connector is present after the latest set of task configs, there's no need to
                // do any zombie fencing or write a new task count record to the config topic
                log.debug("Skipping zombie fencing round for connector {} as all old task generations have already been fenced out", connName);
            } else {
                if (taskCountRecord == null) {
                    // If there is no task count record present for the connector, no transactional producers should have been brought up for it,
                    // so there's nothing to fence--but we do need to write a task count record now so that we know to fence those tasks if/when
                    // the connector is reconfigured
                    log.debug("Skipping zombie fencing round but writing task count record for connector {} "
                            + "as it is being brought up for the first time with exactly-once source support", connName);
                } else {
                    // If the last generation of tasks only had one task, and the next generation only has one, then the new task will automatically
                    // fence out the older task if it's still running; no need to fence here, but again, we still need to write a task count record
                    log.debug("Skipping zombie fencing round but writing task count record for connector {} "
                            + "as both the most recent and the current generation of task configs only contain one task", connName);
                }
                writeToConfigTopicAsLeader(
                        "writing a task count record for connector " + connName + " to the config topic",
                        () -> configBackingStore.putTaskCountRecord(connName, taskCount)
                );
            }
            callback.onCompletion(null, 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