private Void onZombieFencingSuccess() {
            if (!refreshConfigSnapshot(workerSyncTimeoutMs)) {
                throw new ConnectException("Failed to read to end of config topic");
            }
            if (taskGen < configState.taskConfigGeneration(connName)) {
                throw new ConnectRestException(
                    Response.Status.CONFLICT.getStatusCode(),
                    "Fencing failed because new task configurations were generated for the connector");
            }
            // If we've already been cancelled, skip the write to the config topic
            if (fencingFollowup.isDone()) {
                return null;
            }
            writeToConfigTopicAsLeader(
                    "writing a task count record for connector " + connName + " to the config topic",
                    () -> configBackingStore.putTaskCountRecord(connName, tasksToRecord)
            );
            return null;
        }
Start sending requests to the Kafka cluster to fence zombies. In rare cases, may cause blocking calls to
take place before returning, so care should be taken to ensure that this method is not invoked while holding
any important locks (e.g., while synchronized on the surrounding DistributedHerder instance).
This method must be invoked before any {@link #addCallback(Callback) callbacks can be added},
and may only be invoked once.
@throws IllegalStateException if invoked multiple times
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java