@Override
    public void configure(final WorkerConfig config) {
        String topic = config.getString(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG);
        if (topic == null || topic.trim().isEmpty())
            throw new ConfigException("Offset storage topic must be specified");

        this.exactlyOnce = config.exactlyOnceSourceEnabled();

        String clusterId = config.kafkaClusterId();
        String clientId = Objects.requireNonNull(clientIdBase.get()) + "offsets";

        Map<String, Object> originals = config.originals();
        Map<String, Object> producerProps = new HashMap<>(originals);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
        producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.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(CommonClientConfigs.CLIENT_ID_CONFIG, clientId);
        ConnectUtils.addMetricsContextProperties(producerProps, config, clusterId);

        Map<String, Object> consumerProps = new HashMap<>(originals);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.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 offsets topic consumer when exactly-once source support is enabled",
                    false
            );
        }

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

        this.offsetLog = createKafkaBasedLog(topic, producerProps, consumerProps, consumedCallback, topicDescription, topicAdminSupplier, config, Time.SYSTEM);
    }
Create an {@link OffsetBackingStore} backed by a Kafka topic. This constructor will use the given
{@link Supplier} to acquire a {@link TopicAdmin} that will be used for interactions with the backing
Kafka topic. The caller is expected to manage the lifecycle of that object, including
{@link TopicAdmin#close(Duration) closing} it when it is no longer needed.
@param topicAdmin a {@link Supplier} for the {@link TopicAdmin} to use for this backing store;
                  may not be null, and may not return null
@param clientIdBase a {@link Supplier} that will be used to create a
{@link CommonClientConfigs#CLIENT_ID_DOC client ID} for Kafka clients instantiated by this store;
                    may not be null, and may not return null, but may throw {@link UnsupportedOperationException}
                    if this offset store should not create its own Kafka clients
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java