class KafkaAdminClient extends AdminClient {

    @Override
    public AlterConsumerGroupOffsetsResult alterConsumerGroupOffsets(String groupId,
                                                        Map<TopicPartition, OffsetAndMetadata> offsets,
                                                        AlterConsumerGroupOffsetsOptions options) {
        final KafkaFutureImpl<Map<TopicPartition, Errors>> future = new KafkaFutureImpl<>();

        final long startFindCoordinatorMs = time.milliseconds();
        final long deadline = calcDeadlineMs(startFindCoordinatorMs, options.timeoutMs());

        ConsumerGroupOperationContext<Map<TopicPartition, Errors>, AlterConsumerGroupOffsetsOptions> context =
                new ConsumerGroupOperationContext<>(groupId, options, deadline, future);

        Call findCoordinatorCall = getFindCoordinatorCall(context,
            () -> KafkaAdminClient.this.getAlterConsumerGroupOffsetsCall(context, offsets));
        runnable.call(findCoordinatorCall, startFindCoordinatorMs);

        return new AlterConsumerGroupOffsetsResult(future);
    }

    private  KafkaAdminClient(AdminClientConfig config,
                             String clientId,
                             Time time,
                             AdminMetadataManager metadataManager,
                             Metrics metrics,
                             KafkaClient client,
                             TimeoutProcessorFactory timeoutProcessorFactory,
                             LogContext logContext);

}