ConnectorOffsetBackingStore offsetStoreForRegularSourceTask(
            ConnectorTaskId id,
            SourceConnectorConfig sourceConfig,
            Class<? extends Connector> connectorClass,
            Producer<byte[], byte[]> producer,
            Map<String, Object> producerProps,
            TopicAdmin topicAdmin
    ) {
        String connectorSpecificOffsetsTopic = sourceConfig.offsetsTopic();

        if (regularSourceTaskUsesConnectorSpecificOffsetsStore(sourceConfig)) {
            Objects.requireNonNull(topicAdmin, "Source tasks require a non-null topic admin when configured to use their own offsets topic");

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

            KafkaOffsetBackingStore connectorStore =
                    KafkaOffsetBackingStore.readWriteStore(sourceConfig.offsetsTopic(), producer, consumer, topicAdmin, 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, if the user has (implicitly or explicitly) configured the connector with a connector-specific offsets topic
            // but 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 we would not be able to establish reasonable defaults like setting
            // isolation.level=read_committed for the offsets topic consumer for this task
            if (sameOffsetTopicAsWorker(sourceConfig.offsetsTopic(), producerProps)) {
                return ConnectorOffsetBackingStore.withOnlyConnectorStore(
                        () -> LoggingContext.forTask(id),
                        connectorStore,
                        connectorSpecificOffsetsTopic,
                        topicAdmin
                );
            } else {
                return ConnectorOffsetBackingStore.withConnectorAndWorkerStores(
                        () -> LoggingContext.forTask(id),
                        globalOffsetBackingStore,
                        connectorStore,
                        connectorSpecificOffsetsTopic,
                        topicAdmin
                );
            }
        } else {
            return ConnectorOffsetBackingStore.withOnlyWorkerStore(
                    () -> LoggingContext.forTask(id),
                    globalOffsetBackingStore,
                    config.offsetsTopic()
            );
        }
    }
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