KafkaBasedLog<String, byte[]> setupAndCreateKafkaBasedLog(String topic, final WorkerConfig config) {
        String clusterId = config.kafkaClusterId();
        Map<String, Object> originals = config.originals();

        Map<String, Object> producerProps = new HashMap<>(baseProducerProps);
        producerProps.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);

        Map<String, Object> consumerProps = new HashMap<>(originals);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        consumerProps.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);
        ConnectUtils.addMetricsContextProperties(consumerProps, config, clusterId);
        if (config.exactlyOnceSourceEnabled()) {
            ConnectUtils.ensureProperty(
                    consumerProps, ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.toString(),
                    "for the worker's config topic consumer when exactly-once source support is enabled",
                    true
            );
        }

        Map<String, Object> adminProps = new HashMap<>(originals);
        ConnectUtils.addMetricsContextProperties(adminProps, config, clusterId);
        adminProps.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);

        Map<String, Object> topicSettings = config instanceof DistributedConfig
                                            ? ((DistributedConfig) config).configStorageTopicSettings()
                                            : Collections.emptyMap();
        NewTopic topicDescription = TopicAdmin.defineTopic(topic)
                .config(topicSettings) // first so that we override user-supplied settings as needed
                .compacted()
                .partitions(1)
                .replicationFactor(config.getShort(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG))
                .build();

        return createKafkaBasedLog(topic, producerProps, consumerProps, new ConsumeCallback(), topicDescription, topicAdminSupplier, config, time);
    }
@Test
    public void testConsumerPropertiesOverrideUserSuppliedValuesWithExactlyOnceSourceEnabled() {
        props.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "enabled");
        props.put(ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_UNCOMMITTED.toString());
        createStore();

        configStorage.setupAndCreateKafkaBasedLog(TOPIC, config);
        verifyConfigure();

        assertEquals(
                IsolationLevel.READ_COMMITTED.toString(),
                capturedConsumerProps.getValue().get(ISOLATION_LEVEL_CONFIG)
        );
    }