void sinkConnectorOffsets(String connName, Connector connector, Map<String, String> connectorConfig,
                              Callback<ConnectorOffsets> cb) {
        Map<String, Object> adminConfig = adminConfigs(
                connName,
                "connector-worker-adminclient-" + connName,
                config,
                new SinkConnectorConfig(plugins, connectorConfig),
                connector.getClass(),
                connectorClientConfigOverridePolicy,
                kafkaClusterId,
                ConnectorType.SINK);
        String groupId = (String) baseConsumerConfigs(
                connName, "connector-consumer-", config, new SinkConnectorConfig(plugins, connectorConfig),
                connector.getClass(), connectorClientConfigOverridePolicy, kafkaClusterId, ConnectorType.SINK).get(ConsumerConfig.GROUP_ID_CONFIG);
        Admin admin = adminFactory.apply(adminConfig);
        try {
            ListConsumerGroupOffsetsOptions listOffsetsOptions = new ListConsumerGroupOffsetsOptions()
                    .timeoutMs((int) RestServer.DEFAULT_REST_REQUEST_TIMEOUT_MS);
            ListConsumerGroupOffsetsResult listConsumerGroupOffsetsResult = admin.listConsumerGroupOffsets(groupId, listOffsetsOptions);
            listConsumerGroupOffsetsResult.partitionsToOffsetAndMetadata().whenComplete((result, error) -> {
                if (error != null) {
                    log.error("Failed to retrieve consumer group offsets for sink connector {}", connName, error);
                    cb.onCompletion(new ConnectException("Failed to retrieve consumer group offsets for sink connector " + connName, error), null);
                } else {
                    ConnectorOffsets offsets = SinkUtils.consumerGroupOffsetsToConnectorOffsets(result);
                    cb.onCompletion(null, offsets);
                }
                Utils.closeQuietly(admin, "Offset fetch admin for sink connector " + connName);
            });
        } catch (Throwable t) {
            Utils.closeQuietly(admin, "Offset fetch admin for sink connector " + connName);
            log.error("Failed to retrieve consumer group offsets for sink connector {}", connName, t);
            cb.onCompletion(new ConnectException("Failed to retrieve consumer group offsets for sink connector " + connName, t), null);
        }
    }
Get the current consumer group offsets for a sink connector.
<p>
Visible for testing.

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