private void modifySourceConnectorOffsets(String connName, Connector connector, Map<String, String> connectorConfig,
                                              Map<Map<String, ?>, Map<String, ?>> offsets, ClassLoader connectorLoader, Callback<Message> cb) {
        SourceConnectorConfig sourceConfig = new SourceConnectorConfig(plugins, connectorConfig, config.topicCreationEnable());
        Map<String, Object> producerProps = config.exactlyOnceSourceEnabled()
                ? exactlyOnceSourceTaskProducerConfigs(new ConnectorTaskId(connName, 0), config, sourceConfig,
                connector.getClass(), connectorClientConfigOverridePolicy, kafkaClusterId)
                : baseProducerConfigs(connName, "connector-offset-producer-" + connName, config, sourceConfig,
                connector.getClass(), connectorClientConfigOverridePolicy, kafkaClusterId);
        KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(producerProps);

        ConnectorOffsetBackingStore offsetStore = config.exactlyOnceSourceEnabled()
                ? offsetStoreForExactlyOnceSourceConnector(sourceConfig, connName, connector, producer)
                : offsetStoreForRegularSourceConnector(sourceConfig, connName, connector, producer);
        offsetStore.configure(config);

        OffsetStorageWriter offsetWriter = new OffsetStorageWriter(offsetStore, connName, internalKeyConverter, internalValueConverter);
        modifySourceConnectorOffsets(connName, connector, connectorConfig, offsets, offsetStore, producer, offsetWriter, connectorLoader, cb);
    }
Modify (alter / reset) a source connector's offsets.

@param connName the name of the source connector whose offsets are to be modified
@param connector an instance of the source connector
@param connectorConfig the source connector's configuration
@param offsets a mapping from partitions to offsets that need to be written; this should be {@code null} for
               offsets reset requests
@param connectorLoader the connector plugin's classloader to be used as the thread context classloader
@param cb callback to invoke upon completion
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java