private void sourceConnectorOffsets(String connName, Connector connector, Map<String, String> connectorConfig,
                                        Callback<ConnectorOffsets> cb) {
        SourceConnectorConfig sourceConfig = new SourceConnectorConfig(plugins, connectorConfig, config.topicCreationEnable());
        ConnectorOffsetBackingStore offsetStore = config.exactlyOnceSourceEnabled()
                ? offsetStoreForExactlyOnceSourceConnector(sourceConfig, connName, connector, null)
                : offsetStoreForRegularSourceConnector(sourceConfig, connName, connector, null);
        CloseableOffsetStorageReader offsetReader = new OffsetStorageReaderImpl(offsetStore, connName, internalKeyConverter, internalValueConverter);
        sourceConnectorOffsets(connName, offsetStore, offsetReader, cb);
    }
Get the current offsets for a source connector.

@param connName the name of the source connector whose offsets are to be retrieved
@param connector the source connector
@param connectorConfig the source connector's configurations
@param cb callback to invoke upon completion of the request
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java