static Map<String, Object> exactlyOnceSourceTaskProducerConfigs(ConnectorTaskId id,
                                                              WorkerConfig config,
                                                              ConnectorConfig connConfig,
                                                              Class<? extends Connector>  connectorClass,
                                                              ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy,
                                                              String clusterId) {
        Map<String, Object> result = baseProducerConfigs(id.connector(), "connector-producer-" + id, config, connConfig, connectorClass, connectorClientConfigOverridePolicy, clusterId);
        // The base producer properties forcibly disable idempotence; remove it from those properties
        // if not explicitly requested by the user
        boolean connectorProducerIdempotenceConfigured = connConfig.originals().containsKey(
                ConnectorConfig.CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG
        );
        if (!connectorProducerIdempotenceConfigured) {
            boolean workerProducerIdempotenceConfigured = config.originals().containsKey(
                    "producer." + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG
            );
            if (!workerProducerIdempotenceConfigured) {
                result.remove(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG);
            }
        }
        ConnectUtils.ensureProperty(
                result, ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true",
                "for connectors when exactly-once source support is enabled",
                false
        );
        String transactionalId = taskTransactionalId(config.groupId(), id.connector(), id.task());
        ConnectUtils.ensureProperty(
                result, ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId,
                "for connectors when exactly-once source support is enabled",
                true
        );
        return result;
    }
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