public void connectorOffsets(String connName, Map<String, String> connectorConfig, Callback<ConnectorOffsets> cb) {
        String connectorClassOrAlias = connectorConfig.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
        ClassLoader connectorLoader = plugins.connectorLoader(connectorClassOrAlias);

        try (LoaderSwap loaderSwap = plugins.withClassLoader(connectorLoader)) {
            Connector connector = plugins.newConnector(connectorClassOrAlias);
            if (ConnectUtils.isSinkConnector(connector)) {
                log.debug("Fetching offsets for sink connector: {}", connName);
                sinkConnectorOffsets(connName, connector, connectorConfig, cb);
            } else {
                log.debug("Fetching offsets for source connector: {}", connName);
                sourceConnectorOffsets(connName, connector, connectorConfig, cb);
            }
        }
    }
Get the current offsets for a connector. This method is asynchronous and the passed callback is completed when the
request finishes processing.

@param connName the name of the connector whose offsets are to be retrieved
@param connectorConfig the 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