ConnectorOffsetBackingStore offsetStoreForExactlyOnceSourceConnector(
            SourceConnectorConfig sourceConfig,
            String connName,
            Connector connector,
            Producer<byte[], byte[]> producer
    ) {
        String connectorSpecificOffsetsTopic = Optional.ofNullable(sourceConfig.offsetsTopic()).orElse(config.offsetsTopic());

        Map<String, Object> producerProps = baseProducerConfigs(connName, "connector-producer-" + connName, config, sourceConfig, connector.getClass(),
                connectorClientConfigOverridePolicy, kafkaClusterId);

        Map<String, Object> consumerProps = exactlyOnceSourceOffsetsConsumerConfigs(
                    connName, "connector-consumer-" + connName, config, sourceConfig, connector.getClass(),
                    connectorClientConfigOverridePolicy, kafkaClusterId);
        KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerProps);

        Map<String, Object> adminOverrides = adminConfigs(connName, "connector-adminclient-" + connName, config,
                sourceConfig, connector.getClass(), connectorClientConfigOverridePolicy, kafkaClusterId, ConnectorType.SOURCE);

        TopicAdmin admin = new TopicAdmin(adminOverrides);

        KafkaOffsetBackingStore connectorStore = producer == null
                ? KafkaOffsetBackingStore.readOnlyStore(connectorSpecificOffsetsTopic, consumer, admin, internalKeyConverter)
                : KafkaOffsetBackingStore.readWriteStore(connectorSpecificOffsetsTopic, producer, consumer, admin, internalKeyConverter);

        // If the connector's offsets topic is the same as the worker-global offsets topic, there's no need to construct
        // an offset store that has a primary and a secondary store which both read from that same topic.
        // So, even if the user has explicitly configured the connector with a connector-specific offsets topic,
        // if we know that that topic is the same as the worker-global offsets topic, we ignore the worker-global
        // offset store and build a store backed exclusively by a connector-specific offsets store.
        // It may seem reasonable to instead build a store backed exclusively by the worker-global offset store, but that
        // would prevent users from being able to customize the config properties used for the Kafka clients that
        // access the offsets topic, and may lead to confusion for them when tasks are created for the connector
        // since they will all have their own dedicated offsets stores anyways
        if (sameOffsetTopicAsWorker(connectorSpecificOffsetsTopic, producerProps)) {
            return ConnectorOffsetBackingStore.withOnlyConnectorStore(
                    () -> LoggingContext.forConnector(connName),
                    connectorStore,
                    connectorSpecificOffsetsTopic,
                    admin
            );
        } else {
            return ConnectorOffsetBackingStore.withConnectorAndWorkerStores(
                    () -> LoggingContext.forConnector(connName),
                    globalOffsetBackingStore,
                    connectorStore,
                    connectorSpecificOffsetsTopic,
                    admin
            );
        }
    }
Builds and returns an offset backing store for an exactly-once source connector. The offset backing store will either be just
a connector-specific offset backing store (if the connector's offsets topic is the same as the worker's global offset topic)
or a combination of both the worker's global offset backing store and a connector-specific offset backing store.
<p>
Visible for testing.
@param sourceConfig the source connector's config
@param connName the source connector's name
@param connector the source connector
@param producer the Kafka producer for the offset backing store; may be {@code null} if a read-only offset backing store is required
@return An offset backing store for an exactly-once source connector
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java