ConnectorOffsetBackingStore offsetStoreForExactlyOnceSourceTask(
            ConnectorTaskId id,
            SourceConnectorConfig sourceConfig,
            Class<? extends Connector> connectorClass,
            Producer<byte[], byte[]> producer,
            Map<String, Object> producerProps,
            TopicAdmin topicAdmin
    ) {
        Objects.requireNonNull(topicAdmin, "Source tasks require a non-null topic admin when exactly-once support is enabled");

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

        String connectorOffsetsTopic = Optional.ofNullable(sourceConfig.offsetsTopic()).orElse(config.offsetsTopic());

        KafkaOffsetBackingStore connectorStore =
                KafkaOffsetBackingStore.readWriteStore(connectorOffsetsTopic, 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.
        // We cannot under any circumstances build an offset store backed exclusively by the worker-global offset store
        // as that would prevent us from being able to write source records and source offset information for the task
        // with the same producer, and therefore, in the same transaction.
        if (sameOffsetTopicAsWorker(connectorOffsetsTopic, producerProps)) {
            return ConnectorOffsetBackingStore.withOnlyConnectorStore(
                    () -> LoggingContext.forTask(id),
                    connectorStore,
                    connectorOffsetsTopic,
                    topicAdmin
            );
        } else {
            return ConnectorOffsetBackingStore.withConnectorAndWorkerStores(
                    () -> LoggingContext.forTask(id),
                    globalOffsetBackingStore,
                    connectorStore,
                    connectorOffsetsTopic,
                    topicAdmin
            );
        }
    }
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