Map<String, Object> fencableProducerProps(DistributedConfig workerConfig) {
        Map<String, Object> result = new HashMap<>(baseProducerProps(workerConfig));

        result.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-leader");
        // Always require producer acks to all to ensure durable writes
        result.put(ProducerConfig.ACKS_CONFIG, "all");
        // We can set this to 5 instead of 1 without risking reordering because we are using an idempotent producer
        result.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

        ConnectUtils.ensureProperty(
                result, ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true",
                "for the worker's config topic producer when exactly-once source support is enabled or in preparation to be enabled",
                false
        );
        ConnectUtils.ensureProperty(
                result, ProducerConfig.TRANSACTIONAL_ID_CONFIG, workerConfig.transactionalProducerId(),
                "for the worker's config topic producer when exactly-once source support is enabled or in preparation to be enabled",
                true
        );

        return result;
    }
@Test
    public void testFencableProducerPropertiesInsertedByDefault() {
        props.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "preparing");
        String groupId = "my-connect-cluster";
        props.put(GROUP_ID_CONFIG, groupId);
        props.remove(TRANSACTIONAL_ID_CONFIG);
        props.remove(ENABLE_IDEMPOTENCE_CONFIG);
        createStore();

        Map<String, Object> fencableProducerProperties = configStorage.fencableProducerProps(config);
        assertEquals("connect-cluster-" + groupId, fencableProducerProperties.get(TRANSACTIONAL_ID_CONFIG));
        assertEquals("true", fencableProducerProperties.get(ENABLE_IDEMPOTENCE_CONFIG));
    }