private void alterSinkConnectorOffsets(String connName, String groupId, Admin admin, Map<TopicPartition, Long> offsetsToWrite,
                                           Callback<Message> cb, boolean alterOffsetsResult, Timer timer) {
        List<KafkaFuture<Void>> adminFutures = new ArrayList<>();

        Map<TopicPartition, OffsetAndMetadata> offsetsToAlter = offsetsToWrite.entrySet()
                .stream()
                .filter(entry -> entry.getValue() != null)
                .collect(Collectors.toMap(Map.Entry::getKey, e -> new OffsetAndMetadata(e.getValue())));

        if (!offsetsToAlter.isEmpty()) {
            log.debug("Committing the following consumer group offsets using an admin client for sink connector {}: {}.",
                    connName, offsetsToAlter);
            AlterConsumerGroupOffsetsOptions alterConsumerGroupOffsetsOptions = new AlterConsumerGroupOffsetsOptions()
                    .timeoutMs((int) timer.remainingMs());
            AlterConsumerGroupOffsetsResult alterConsumerGroupOffsetsResult = admin.alterConsumerGroupOffsets(groupId, offsetsToAlter,
                    alterConsumerGroupOffsetsOptions);

            adminFutures.add(alterConsumerGroupOffsetsResult.all());
        }

        Set<TopicPartition> partitionsToReset = offsetsToWrite.entrySet()
                .stream()
                .filter(entry -> entry.getValue() == null)
                .map(Map.Entry::getKey)
                .collect(Collectors.toSet());

        if (!partitionsToReset.isEmpty()) {
            log.debug("Deleting the consumer group offsets for the following topic partitions using an admin client for sink connector {}: {}.",
                    connName, partitionsToReset);
            DeleteConsumerGroupOffsetsOptions deleteConsumerGroupOffsetsOptions = new DeleteConsumerGroupOffsetsOptions()
                    .timeoutMs((int) timer.remainingMs());
            DeleteConsumerGroupOffsetsResult deleteConsumerGroupOffsetsResult = admin.deleteConsumerGroupOffsets(groupId, partitionsToReset,
                    deleteConsumerGroupOffsetsOptions);

            adminFutures.add(deleteConsumerGroupOffsetsResult.all());
        }

        @SuppressWarnings("rawtypes")
        KafkaFuture<Void> compositeAdminFuture = KafkaFuture.allOf(adminFutures.toArray(new KafkaFuture[0]));

        compositeAdminFuture.whenComplete((ignored, error) -> {
            if (error != null) {
                // When a consumer group is non-empty, only group members can commit offsets. An attempt to alter offsets via the admin client
                // will result in an UnknownMemberIdException if the consumer group is non-empty (i.e. if the sink tasks haven't stopped
                // completely or if the connector is resumed while the alter offsets request is being processed). Similarly, an attempt to
                // delete consumer group offsets for a non-empty consumer group will result in a GroupSubscribedToTopicException
                if (error instanceof UnknownMemberIdException || error instanceof GroupSubscribedToTopicException) {
                    String errorMsg = "Failed to alter consumer group offsets for connector " + connName + " either because its tasks " +
                            "haven't stopped completely yet or the connector was resumed before the request to alter its offsets could be successfully " +
                            "completed. If the connector is in a stopped state, this operation can be safely retried. If it doesn't eventually succeed, the " +
                            "Connect cluster may need to be restarted to get rid of the zombie sink tasks.";
                    log.error(errorMsg, error);
                    cb.onCompletion(new ConnectException(errorMsg, error), null);
                } else {
                    log.error("Failed to alter consumer group offsets for connector {}", connName, error);
                    cb.onCompletion(new ConnectException("Failed to alter consumer group offsets for connector " + connName, error), null);
                }
            } else {
                completeModifyOffsetsCallback(alterOffsetsResult, false, cb);
            }
        }).whenComplete((ignored, ignoredError) -> {
            // errors originating from the original future are handled in the prior whenComplete invocation which isn't expected to throw
            // an exception itself, and we can thus ignore the error here
            Utils.closeQuietly(admin, "Offset alter admin for sink connector " + connName);
        });
    }
Alter a sink connector's consumer group offsets. This is done via calls to {@link Admin#alterConsumerGroupOffsets}
and / or {@link Admin#deleteConsumerGroupOffsets}.

@param connName the name of the sink connector whose offsets are to be altered
@param groupId the sink connector's consumer group ID
@param admin the {@link Admin admin client} to be used for altering the consumer group offsets; will be closed after use
@param offsetsToWrite a mapping from topic partitions to offsets that need to be written; may not be null or empty
@param cb callback to invoke upon completion
@param alterOffsetsResult the result of the call to {@link SinkConnector#alterOffsets} for the connector
@param timer {@link Timer} to bound the total runtime of admin client requests
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java