static Map<String, Object> baseProducerConfigs(String connName,
                                               String defaultClientId,
                                               WorkerConfig config,
                                               ConnectorConfig connConfig,
                                               Class<? extends Connector>  connectorClass,
                                               ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy,
                                               String clusterId) {
        Map<String, Object> producerProps = new HashMap<>();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.bootstrapServers());
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
        // These settings will execute infinite retries on retriable exceptions. They *may* be overridden via configs passed to the worker,
        // but this may compromise the delivery guarantees of Kafka Connect.
        producerProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, Long.toString(Long.MAX_VALUE));
        // By default, Connect disables idempotent behavior for all producers, even though idempotence became
        // default for Kafka producers. This is to ensure Connect continues to work with many Kafka broker versions, including older brokers that do not support
        // idempotent producers or require explicit steps to enable them (e.g. adding the IDEMPOTENT_WRITE ACL to brokers older than 2.8).
        // These settings might change when https://cwiki.apache.org/confluence/display/KAFKA/KIP-318%3A+Make+Kafka+Connect+Source+idempotent
        // gets approved and scheduled for release.
        producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");
        producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
        producerProps.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1");
        producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE));
        producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, defaultClientId);
        // User-specified overrides
        producerProps.putAll(config.originalsWithPrefix("producer."));
        //add client metrics.context properties
        ConnectUtils.addMetricsContextProperties(producerProps, config, clusterId);

        // Connector-specified overrides
        Map<String, Object> producerOverrides =
            connectorClientConfigOverrides(connName, connConfig, connectorClass, ConnectorConfig.CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX,
                                           ConnectorType.SOURCE, ConnectorClientConfigRequest.ClientType.PRODUCER,
                                           connectorClientConfigOverridePolicy);
        producerProps.putAll(producerOverrides);

        return producerProps;
    }
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