void modifySourceConnectorOffsets(String connName, Connector connector, Map<String, String> connectorConfig,
                                      Map<Map<String, ?>, Map<String, ?>> offsets, ConnectorOffsetBackingStore offsetStore,
                                      KafkaProducer<byte[], byte[]> producer, OffsetStorageWriter offsetWriter,
                                      ClassLoader connectorLoader, Callback<Message> cb) {
        executor.submit(plugins.withClassLoader(connectorLoader, () -> {
            try {
                Timer timer = time.timer(Duration.ofMillis(RestServer.DEFAULT_REST_REQUEST_TIMEOUT_MS));
                // This reads to the end of the offsets topic and can be a potentially time-consuming operation
                offsetStore.start();
                updateTimerAndCheckExpiry(timer, "Timed out while trying to read to the end of the offsets topic prior to modifying " +
                        "offsets for source connector " + connName);
                Map<Map<String, ?>, Map<String, ?>> offsetsToWrite;

                // If the offsets argument is null, it indicates an offsets reset operation - i.e. a null offset should
                // be written for every source partition of the connector
                boolean isReset;
                if (offsets == null) {
                    isReset = true;
                    offsetsToWrite = new HashMap<>();
                    offsetStore.connectorPartitions(connName).forEach(partition -> offsetsToWrite.put(partition, null));
                    log.debug("Found the following partitions (to reset offsets) for source connector {}: {}", connName, offsetsToWrite.keySet());
                } else {
                    isReset = false;
                    offsetsToWrite = offsets;
                }

                Map<Map<String, ?>, Map<String, ?>> normalizedOffsets = normalizeSourceConnectorOffsets(offsetsToWrite);

                boolean alterOffsetsResult;
                try {
                    alterOffsetsResult = ((SourceConnector) connector).alterOffsets(connectorConfig, normalizedOffsets);
                } 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 source connector " + connName);

                // This should only occur for an offsets reset request when there are no source partitions found for the source connector in the
                // offset store - either because there was a prior attempt to reset offsets or if there are no offsets committed by this source
                // connector so far
                if (normalizedOffsets.isEmpty()) {
                    log.info("No offsets found for source connector {} - this can occur due to a prior attempt to reset offsets or if the " +
                            "source connector hasn't committed any offsets yet", connName);
                    completeModifyOffsetsCallback(alterOffsetsResult, isReset, cb);
                    return;
                }

                // The modifySourceConnectorOffsets method should only be called after all the connector's tasks have been stopped, and it's
                // safe to write offsets via an offset writer
                normalizedOffsets.forEach(offsetWriter::offset);

                // We can call begin flush without a timeout because this newly created single-purpose offset writer can't do concurrent
                // offset writes. We can also ignore the return value since it returns false if and only if there is no data to be flushed,
                // and we've just put some data in the previous statement
                offsetWriter.beginFlush();

                if (config.exactlyOnceSourceEnabled()) {
                    producer.initTransactions();
                    producer.beginTransaction();
                }
                log.debug("Committing the following offsets for source connector {}: {}", connName, normalizedOffsets);
                FutureCallback<Void> offsetWriterCallback = new FutureCallback<>();
                offsetWriter.doFlush(offsetWriterCallback);
                if (config.exactlyOnceSourceEnabled()) {
                    producer.commitTransaction();
                }

                try {
                    offsetWriterCallback.get(timer.remainingMs(), TimeUnit.MILLISECONDS);
                } catch (ExecutionException e) {
                    throw new ConnectException("Failed to modify offsets for source connector " + connName, e.getCause());
                } catch (TimeoutException e) {
                    throw new ConnectException("Timed out while attempting to modify offsets for source connector " + connName, e);
                } catch (InterruptedException e) {
                    throw new ConnectException("Unexpectedly interrupted while attempting to modify offsets for source connector " + connName, e);
                }

                completeModifyOffsetsCallback(alterOffsetsResult, isReset, cb);
            } catch (Throwable t) {
                log.error("Failed to modify offsets for source connector {}", connName, t);
                cb.onCompletion(ConnectUtils.maybeWrap(t, "Failed to modify offsets for source connector " + connName), null);
            } finally {
                Utils.closeQuietly(offsetStore::stop, "Offset store for offset modification request for connector " + connName);
            }
        }));
    }
Modify (alter / reset) a source connector's offsets.

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