public void modifyConnectorOffsets(String connName, Map<String, String> connectorConfig,
                                      Map<Map<String, ?>, Map<String, ?>> offsets, Callback<Message> cb) {
        String connectorClassOrAlias = connectorConfig.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
        ClassLoader connectorLoader = plugins.connectorLoader(connectorClassOrAlias);
        Connector connector;

        try (LoaderSwap loaderSwap = plugins.withClassLoader(connectorLoader)) {
            connector = plugins.newConnector(connectorClassOrAlias);
            if (ConnectUtils.isSinkConnector(connector)) {
                log.debug("Modifying offsets for sink connector: {}", connName);
                modifySinkConnectorOffsets(connName, connector, connectorConfig, offsets, connectorLoader, cb);
            } else {
                log.debug("Modifying offsets for source connector: {}", connName);
                modifySourceConnectorOffsets(connName, connector, connectorConfig, offsets, connectorLoader, cb);
            }
        }
    }
Modify (alter / reset) a connector's offsets.

@param connName the name of the connector whose offsets are to be modified
@param connectorConfig the connector's configurations
@param offsets a mapping from partitions (either source partitions for source connectors, or Kafka topic
               partitions for sink connectors) to offsets that need to be written; this should be {@code null}
               for offsets reset requests
@param cb callback to invoke upon completion
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java