ConnectorOffsetBackingStore offsetStoreForRegularSourceConnector(
            SourceConnectorConfig sourceConfig,
            String connName,
            Connector connector,
            Producer<byte[], byte[]> producer
    ) {
        String connectorSpecificOffsetsTopic = sourceConfig.offsetsTopic();

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

        // We use a connector-specific store (i.e., a dedicated KafkaOffsetBackingStore for this connector)
        // if the worker supports per-connector offsets topics (which may be the case in distributed but not standalone mode, for example)
        // and if the connector is explicitly configured with an offsets topic
        final boolean usesConnectorSpecificStore = connectorSpecificOffsetsTopic != null
                && config.connectorOffsetsTopicsPermitted();

        if (usesConnectorSpecificStore) {
            Map<String, Object> consumerProps = regularSourceOffsetsConsumerConfigs(
                        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, if the user has 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 connector
            if (sameOffsetTopicAsWorker(connectorSpecificOffsetsTopic, producerProps)) {
                return ConnectorOffsetBackingStore.withOnlyConnectorStore(
                        () -> LoggingContext.forConnector(connName),
                        connectorStore,
                        connectorSpecificOffsetsTopic,
                        admin
                );
            } else {
                return ConnectorOffsetBackingStore.withConnectorAndWorkerStores(
                        () -> LoggingContext.forConnector(connName),
                        globalOffsetBackingStore,
                        connectorStore,
                        connectorSpecificOffsetsTopic,
                        admin
                );
            }
        } else {
            Utils.closeQuietly(producer, "Unused producer for offset store");
            return ConnectorOffsetBackingStore.withOnlyWorkerStore(
                    () -> LoggingContext.forConnector(connName),
                    globalOffsetBackingStore,
                    config.offsetsTopic()
            );
        }
    }
Builds and returns an offset backing store for a regular source connector (i.e. when exactly-once support for source connectors is disabled).
The offset backing store will either be just the worker's global offset backing store (if the connector doesn't define a connector-specific
offset topic via its configs), just a connector-specific offset backing store (if the connector defines a connector-specific offsets
topic which appears to be 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 a regular source connector
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java