public KafkaFuture<Void> fenceZombies(String connName, int numTasks, Map<String, String> connProps) {
        log.debug("Fencing out {} task producers for source connector {}", numTasks, connName);
        try (LoggingContext loggingContext = LoggingContext.forConnector(connName)) {
            String connType = connProps.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
            ClassLoader connectorLoader = plugins.connectorLoader(connType);
            try (LoaderSwap loaderSwap = plugins.withClassLoader(connectorLoader)) {
                final SourceConnectorConfig connConfig = new SourceConnectorConfig(plugins, connProps, config.topicCreationEnable());
                final Class<? extends Connector> connClass = plugins.connectorClass(
                        connConfig.getString(ConnectorConfig.CONNECTOR_CLASS_CONFIG));

                Map<String, Object> adminConfig = adminConfigs(
                        connName,
                        "connector-worker-adminclient-" + connName,
                        config,
                        connConfig,
                        connClass,
                        connectorClientConfigOverridePolicy,
                        kafkaClusterId,
                        ConnectorType.SOURCE);
                final Admin admin = adminFactory.apply(adminConfig);

                try {
                    Collection<String> transactionalIds = IntStream.range(0, numTasks)
                            .mapToObj(i -> new ConnectorTaskId(connName, i))
                            .map(this::taskTransactionalId)
                            .collect(Collectors.toList());
                    FenceProducersOptions fencingOptions = new FenceProducersOptions()
                            .timeoutMs((int) RestServer.DEFAULT_REST_REQUEST_TIMEOUT_MS);
                    return admin.fenceProducers(transactionalIds, fencingOptions).all().whenComplete((ignored, error) -> {
                        if (error == null)
                            log.debug("Finished fencing out {} task producers for source connector {}", numTasks, connName);
                        Utils.closeQuietly(admin, "Zombie fencing admin for connector " + connName);
                    });
                } catch (Exception e) {
                    Utils.closeQuietly(admin, "Zombie fencing admin for connector " + connName);
                    throw e;
                }
            }
        }
    }
Using the admin principal for this connector, perform a round of zombie fencing that disables transactional producers
for the specified number of source tasks from sending any more records.
@param connName the name of the connector
@param numTasks the number of tasks to fence out
@param connProps the configuration of the connector; may not be null
@return a {@link KafkaFuture} that will complete when the producers have all been fenced out, or the attempt has failed
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java