void sourceConnectorOffsets(String connName, ConnectorOffsetBackingStore offsetStore,
                                CloseableOffsetStorageReader offsetReader, Callback<ConnectorOffsets> cb) {
        executor.submit(() -> {
            try {
                offsetStore.configure(config);
                offsetStore.start();
                Set<Map<String, Object>> connectorPartitions = offsetStore.connectorPartitions(connName);
                List<ConnectorOffset> connectorOffsets = offsetReader.offsets(connectorPartitions).entrySet().stream()
                        .map(entry -> new ConnectorOffset(entry.getKey(), entry.getValue()))
                        .collect(Collectors.toList());
                cb.onCompletion(null, new ConnectorOffsets(connectorOffsets));
            } catch (Throwable t) {
                log.error("Failed to retrieve offsets for source connector {}", connName, t);
                cb.onCompletion(ConnectUtils.maybeWrap(t, "Failed to retrieve offsets for source connector " + connName), null);
            } finally {
                Utils.closeQuietly(offsetReader, "Offset reader for connector " + connName);
                Utils.closeQuietly(offsetStore::stop, "Offset store for connector " + connName);
            }
        });
    }
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