private void resetSinkConnectorOffsets(String connName, String groupId, Admin admin, Callback<Message> cb, boolean alterOffsetsResult, Timer timer) {
        DeleteConsumerGroupsOptions deleteConsumerGroupsOptions = new DeleteConsumerGroupsOptions().timeoutMs((int) timer.remainingMs());

        admin.deleteConsumerGroups(Collections.singleton(groupId), deleteConsumerGroupsOptions)
                .all()
                .whenComplete((ignored, error) -> {
                    // We treat GroupIdNotFoundException as a non-error here because resetting a connector's offsets is expected to be an idempotent operation
                    // and the consumer group could have already been deleted in a prior offsets reset request
                    if (error != null && !(error instanceof GroupIdNotFoundException)) {
                        // When a consumer group is non-empty, attempts to delete it via the admin client result in a GroupNotEmptyException. This can occur
                        // if the sink tasks haven't stopped completely or if the connector is resumed while the reset offsets request is being processed
                        if (error instanceof GroupNotEmptyException) {
                            String errorMsg = "Failed to reset consumer group offsets for connector " + connName + " either because its tasks " +
                                    "haven't stopped completely yet or the connector was resumed before the request to reset 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 reset consumer group offsets for sink connector {}", connName, error);
                            cb.onCompletion(new ConnectException("Failed to reset consumer group offsets for sink connector " + connName, error), null);
                        }
                    } else {
                        completeModifyOffsetsCallback(alterOffsetsResult, true, 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 reset admin for sink connector " + connName);
                });
    }
Reset a sink connector's consumer group offsets. This is done by deleting the consumer group via a call to
{@link Admin#deleteConsumerGroups}

@param connName the name of the sink connector whose offsets are to be reset
@param groupId the sink connector's consumer group ID
@param admin the {@link Admin admin client} to be used for resetting the consumer group offsets; will be closed after use
@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