void modifySinkConnectorOffsets(String connName, Connector connector, Map<String, String> connectorConfig,
                                    Map<Map<String, ?>, Map<String, ?>> offsets, ClassLoader connectorLoader, Callback<Message> cb) {
        executor.submit(plugins.withClassLoader(connectorLoader, () -> {
            try {
                Timer timer = time.timer(Duration.ofMillis(RestServer.DEFAULT_REST_REQUEST_TIMEOUT_MS));
                boolean isReset = offsets == null;
                SinkConnectorConfig sinkConnectorConfig = new SinkConnectorConfig(plugins, connectorConfig);
                Class<? extends Connector> sinkConnectorClass = connector.getClass();
                Map<String, Object> adminConfig = adminConfigs(
                        connName,
                        "connector-worker-adminclient-" + connName,
                        config,
                        sinkConnectorConfig,
                        sinkConnectorClass,
                        connectorClientConfigOverridePolicy,
                        kafkaClusterId,
                        ConnectorType.SINK);

                String groupId = (String) baseConsumerConfigs(
                        connName, "connector-consumer-", config, sinkConnectorConfig,
                        sinkConnectorClass, connectorClientConfigOverridePolicy, kafkaClusterId, ConnectorType.SINK).get(ConsumerConfig.GROUP_ID_CONFIG);

                Admin admin = adminFactory.apply(adminConfig);

                try {
                    Map<TopicPartition, Long> offsetsToWrite;
                    if (isReset) {
                        offsetsToWrite = new HashMap<>();
                        ListConsumerGroupOffsetsOptions listConsumerGroupOffsetsOptions = new ListConsumerGroupOffsetsOptions()
                                .timeoutMs((int) timer.remainingMs());
                        try {
                            admin.listConsumerGroupOffsets(groupId, listConsumerGroupOffsetsOptions)
                                    .partitionsToOffsetAndMetadata()
                                    .get(timer.remainingMs(), TimeUnit.MILLISECONDS)
                                    .forEach((topicPartition, offsetAndMetadata) -> offsetsToWrite.put(topicPartition, null));

                            timer.update();
                            log.debug("Found the following topic partitions (to reset offsets) for sink connector {} and consumer group ID {}: {}",
                                    connName, groupId, offsetsToWrite.keySet());
                        } catch (Exception e) {
                            Utils.closeQuietly(admin, "Offset reset admin for sink connector " + connName);
                            log.error("Failed to list offsets prior to resetting offsets for sink connector {}", connName, e);
                            cb.onCompletion(new ConnectException("Failed to list offsets prior to resetting offsets for sink connector " + connName, e), null);
                            return;
                        }
                    } else {
                        offsetsToWrite = SinkUtils.parseSinkConnectorOffsets(offsets);
                    }

                    boolean alterOffsetsResult;
                    try {
                        alterOffsetsResult = ((SinkConnector) connector).alterOffsets(connectorConfig, offsetsToWrite);
                    } catch (UnsupportedOperationException e) {
                        log.error("Failed to modify offsets for connector {} because it doesn't support external modification of offsets",
                                connName, e);
                        throw new ConnectException("Failed to modify offsets for connector " + connName + " because it doesn't support external " +
                                "modification of offsets", e);
                    }
                    updateTimerAndCheckExpiry(timer, "Timed out while calling the 'alterOffsets' method for sink connector " + connName);

                    if (isReset) {
                        resetSinkConnectorOffsets(connName, groupId, admin, cb, alterOffsetsResult, timer);
                    } else {
                        alterSinkConnectorOffsets(connName, groupId, admin, offsetsToWrite, cb, alterOffsetsResult, timer);
                    }
                } catch (Throwable t) {
                    Utils.closeQuietly(admin, "Offset modification admin for sink connector " + connName);
                    throw t;
                }
            } catch (Throwable t) {
                log.error("Failed to modify offsets for sink connector {}", connName, t);
                cb.onCompletion(ConnectUtils.maybeWrap(t, "Failed to modify offsets for sink connector " + connName), null);
            }
        }));
    }
Modify (alter / reset) a sink connector's consumer group offsets.
<p>
Visible for testing.

@param connName the name of the sink connector whose offsets are to be modified
@param connector an instance of the sink connector
@param connectorConfig the sink connector's configuration
@param offsets a mapping from topic 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