content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.coordinator.group;
  
  import org.apache.kafka.common.TopicPartition;
  import org.apache.kafka.common.Uuid;
  import org.apache.kafka.common.compress.Compression;
  import org.apache.kafka.common.config.TopicConfig;
  import org.apache.kafka.common.errors.InvalidRequestException;
  import org.apache.kafka.common.errors.NotCoordinatorException;
  import org.apache.kafka.common.errors.StreamsInvalidTopologyException;
  import org.apache.kafka.common.errors.UnsupportedAssignorException;
  import org.apache.kafka.common.internals.Plugin;
  import org.apache.kafka.common.internals.Topic;
  import org.apache.kafka.common.message.AlterShareGroupOffsetsRequestData;
  import org.apache.kafka.common.message.AlterShareGroupOffsetsResponseData;
  import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
  import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
  import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
  import org.apache.kafka.common.message.DeleteGroupsResponseData;
  import org.apache.kafka.common.message.DeleteShareGroupOffsetsRequestData;
  import org.apache.kafka.common.message.DeleteShareGroupOffsetsResponseData;
  import org.apache.kafka.common.message.DescribeGroupsResponseData;
  import org.apache.kafka.common.message.DescribeShareGroupOffsetsRequestData;
  import org.apache.kafka.common.message.DescribeShareGroupOffsetsResponseData;
  import org.apache.kafka.common.message.HeartbeatRequestData;
  import org.apache.kafka.common.message.HeartbeatResponseData;
  import org.apache.kafka.common.message.JoinGroupRequestData;
  import org.apache.kafka.common.message.JoinGroupResponseData;
  import org.apache.kafka.common.message.LeaveGroupRequestData;
  import org.apache.kafka.common.message.LeaveGroupResponseData;
  import org.apache.kafka.common.message.ListGroupsRequestData;
  import org.apache.kafka.common.message.ListGroupsResponseData;
  import org.apache.kafka.common.message.OffsetCommitRequestData;
  import org.apache.kafka.common.message.OffsetCommitResponseData;
  import org.apache.kafka.common.message.OffsetDeleteRequestData;
  import org.apache.kafka.common.message.OffsetDeleteResponseData;
  import org.apache.kafka.common.message.OffsetFetchRequestData;
  import org.apache.kafka.common.message.OffsetFetchResponseData;
  import org.apache.kafka.common.message.ReadShareGroupStateSummaryRequestData;
  import org.apache.kafka.common.message.ShareGroupDescribeResponseData;
  import org.apache.kafka.common.message.ShareGroupDescribeResponseData.DescribedGroup;
  import org.apache.kafka.common.message.ShareGroupHeartbeatRequestData;
  import org.apache.kafka.common.message.ShareGroupHeartbeatResponseData;
  import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
  import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
  import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
  import org.apache.kafka.common.message.SyncGroupRequestData;
  import org.apache.kafka.common.message.SyncGroupResponseData;
  import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
  import org.apache.kafka.common.message.TxnOffsetCommitResponseData;
  import org.apache.kafka.common.protocol.Errors;
  import org.apache.kafka.common.requests.AlterShareGroupOffsetsRequest;
  import org.apache.kafka.common.requests.ApiError;
  import org.apache.kafka.common.requests.ConsumerGroupDescribeRequest;
  import org.apache.kafka.common.requests.DeleteGroupsRequest;
  import org.apache.kafka.common.requests.DeleteShareGroupOffsetsRequest;
  import org.apache.kafka.common.requests.DescribeGroupsRequest;
  import org.apache.kafka.common.requests.DescribeShareGroupOffsetsRequest;
  import org.apache.kafka.common.requests.OffsetCommitRequest;
  import org.apache.kafka.common.requests.ShareGroupDescribeRequest;
  import org.apache.kafka.common.requests.ShareGroupHeartbeatRequest;
  import org.apache.kafka.common.requests.StreamsGroupDescribeRequest;
  import org.apache.kafka.common.requests.TransactionResult;
  import org.apache.kafka.common.requests.TxnOffsetCommitRequest;
  import org.apache.kafka.common.utils.BufferSupplier;
  import org.apache.kafka.common.utils.LogContext;
  import org.apache.kafka.common.utils.Time;
  import org.apache.kafka.common.utils.Utils;
  import org.apache.kafka.coordinator.common.runtime.CoordinatorEventProcessor;
  import org.apache.kafka.coordinator.common.runtime.CoordinatorLoader;
  import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
  import org.apache.kafka.coordinator.common.runtime.CoordinatorResult;
  import org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime;
  import org.apache.kafka.coordinator.common.runtime.CoordinatorRuntimeMetrics;
  import org.apache.kafka.coordinator.common.runtime.CoordinatorShardBuilderSupplier;
  import org.apache.kafka.coordinator.common.runtime.MultiThreadedEventProcessor;
  import org.apache.kafka.coordinator.common.runtime.PartitionWriter;
  import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
  import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
  import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
  import org.apache.kafka.image.MetadataDelta;
  import org.apache.kafka.image.MetadataImage;
  import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
  import org.apache.kafka.server.authorizer.Authorizer;
  import org.apache.kafka.server.record.BrokerCompressionType;
  import org.apache.kafka.server.share.persister.DeleteShareGroupStateParameters;
  import org.apache.kafka.server.share.persister.DeleteShareGroupStateResult;
  import org.apache.kafka.server.share.persister.GroupTopicPartitionData;
  import org.apache.kafka.server.share.persister.InitializeShareGroupStateParameters;
  import org.apache.kafka.server.share.persister.InitializeShareGroupStateResult;
  import org.apache.kafka.server.share.persister.PartitionErrorData;
  import org.apache.kafka.server.share.persister.PartitionFactory;
  import org.apache.kafka.server.share.persister.PartitionStateData;
  import org.apache.kafka.server.share.persister.Persister;
  import org.apache.kafka.server.share.persister.ReadShareGroupStateSummaryParameters;
  import org.apache.kafka.server.share.persister.TopicData;
  import org.apache.kafka.server.util.FutureUtils;
  import org.apache.kafka.server.util.timer.Timer;
  import org.apache.kafka.server.util.timer.TimerTask;
  
  import org.slf4j.Logger;
  
  import java.time.Duration;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  import java.util.OptionalInt;
  import java.util.Properties;
  import java.util.Set;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.CompletionException;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.Executors;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.function.IntSupplier;
  import java.util.stream.Collectors;
  
  import static org.apache.kafka.common.requests.ConsumerGroupHeartbeatRequest.CONSUMER_GENERATED_MEMBER_ID_REQUIRED_VERSION;
  import static org.apache.kafka.common.requests.ConsumerGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH;
  import static org.apache.kafka.common.requests.ConsumerGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH;
  import static org.apache.kafka.coordinator.common.runtime.CoordinatorOperationExceptionHelper.handleOperationException;
  import static org.apache.kafka.coordinator.group.Utils.throwIfEmptyString;
  import static org.apache.kafka.coordinator.group.Utils.throwIfNotEmptyCollection;
  import static org.apache.kafka.coordinator.group.Utils.throwIfNotNull;
  import static org.apache.kafka.coordinator.group.Utils.throwIfNull;
  
  /**
   * The group coordinator service.
   */
  @SuppressWarnings({"ClassDataAbstractionCoupling"})
  public class GroupCoordinatorService implements GroupCoordinator {
  
      public static class Builder {
          private final int nodeId;
          private final GroupCoordinatorConfig config;
          private PartitionWriter writer;
          private CoordinatorLoader<CoordinatorRecord> loader;
          private Time time;
          private Timer timer;
          private CoordinatorRuntimeMetrics coordinatorRuntimeMetrics;
          private GroupCoordinatorMetrics groupCoordinatorMetrics;
          private GroupConfigManager groupConfigManager;
          private Persister persister;
          private Optional<Plugin<Authorizer>> authorizerPlugin;
  
          public Builder(
              int nodeId,
              GroupCoordinatorConfig config
          ) {
              this.nodeId = nodeId;
              this.config = config;
          }
  
          public Builder withWriter(PartitionWriter writer) {
              this.writer = writer;
              return this;
          }
  
          public Builder withLoader(CoordinatorLoader<CoordinatorRecord> loader) {
              this.loader = loader;
              return this;
          }
  
          public Builder withTime(Time time) {
              this.time = time;
              return this;
          }
  
          public Builder withTimer(Timer timer) {
              this.timer = timer;
              return this;
          }
  
          public Builder withCoordinatorRuntimeMetrics(CoordinatorRuntimeMetrics coordinatorRuntimeMetrics) {
              this.coordinatorRuntimeMetrics = coordinatorRuntimeMetrics;
              return this;
          }
  
          public Builder withGroupCoordinatorMetrics(GroupCoordinatorMetrics groupCoordinatorMetrics) {
              this.groupCoordinatorMetrics = groupCoordinatorMetrics;
              return this;
          }
  
          public Builder withGroupConfigManager(GroupConfigManager groupConfigManager) {
              this.groupConfigManager = groupConfigManager;
              return this;
          }
  
          public Builder withPersister(Persister persister) {
              this.persister = persister;
              return this;
          }
  
          public Builder withAuthorizerPlugin(Optional<Plugin<Authorizer>> authorizerPlugin) {
              this.authorizerPlugin = authorizerPlugin;
              return this;
          }
  
          public GroupCoordinatorService build() {
              requireNonNull(config, "Config must be set.");
              requireNonNull(writer, "Writer must be set.");
              requireNonNull(loader, "Loader must be set.");
              requireNonNull(time, "Time must be set.");
              requireNonNull(timer, "Timer must be set.");
              requireNonNull(coordinatorRuntimeMetrics, "CoordinatorRuntimeMetrics must be set.");
              requireNonNull(groupCoordinatorMetrics, "GroupCoordinatorMetrics must be set.");
              requireNonNull(groupConfigManager, "GroupConfigManager must be set.");
              requireNonNull(persister, "Persister must be set.");
              requireNonNull(authorizerPlugin, "Authorizer must be set.");
  
              String logPrefix = String.format("GroupCoordinator id=%d", nodeId);
              LogContext logContext = new LogContext(String.format("[%s] ", logPrefix));
  
              CoordinatorShardBuilderSupplier<GroupCoordinatorShard, CoordinatorRecord> supplier = () ->
                  new GroupCoordinatorShard.Builder(config, groupConfigManager)
                      .withAuthorizerPlugin(authorizerPlugin);
  
              CoordinatorEventProcessor processor = new MultiThreadedEventProcessor(
                  logContext,
                  "group-coordinator-event-processor-",
                  config.numThreads(),
                  time,
                  coordinatorRuntimeMetrics
              );
  
              CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
                  new CoordinatorRuntime.Builder<GroupCoordinatorShard, CoordinatorRecord>()
                      .withTime(time)
                      .withTimer(timer)
                      .withLogPrefix(logPrefix)
                      .withLogContext(logContext)
                      .withEventProcessor(processor)
                      .withPartitionWriter(writer)
                      .withLoader(loader)
                      .withCoordinatorShardBuilderSupplier(supplier)
                      .withDefaultWriteTimeOut(Duration.ofMillis(config.offsetCommitTimeoutMs()))
                      .withCoordinatorRuntimeMetrics(coordinatorRuntimeMetrics)
                      .withCoordinatorMetrics(groupCoordinatorMetrics)
                      .withSerializer(new GroupCoordinatorRecordSerde())
                      .withCompression(Compression.of(config.offsetTopicCompressionType()).build())
                      .withAppendLingerMs(config.appendLingerMs())
                      .withExecutorService(Executors.newSingleThreadExecutor())
                      .build();
  
              return new GroupCoordinatorService(
                  logContext,
                  config,
                  runtime,
                  groupCoordinatorMetrics,
                  groupConfigManager,
                  persister,
                  timer
              );
          }
      }
  
      /**
       * The logger.
       */
      private final Logger log;
  
      /**
       * The group coordinator configurations.
       */
      private final GroupCoordinatorConfig config;
  
      /**
       * The coordinator runtime.
       */
      private final CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime;
  
      /**
       * The metrics registry.
       */
      private final GroupCoordinatorMetrics groupCoordinatorMetrics;
  
      /**
       * The group config manager.
       */
      private final GroupConfigManager groupConfigManager;
  
      /**
       * The Persister to persist the state of share partition state.
       */
      private final Persister persister;
  
      /**
       * The timer to schedule tasks.
       */
      private final Timer timer;
  
      /**
       * Boolean indicating whether the coordinator is active or not.
       */
      private final AtomicBoolean isActive = new AtomicBoolean(false);
  
      /**
       * The set of supported consumer group assignors.
       */
      private final Set<String> consumerGroupAssignors;
  
      /**
       * The number of partitions of the __consumer_offsets topics. This is provided
       * when the component is started.
       */
      private volatile int numPartitions = -1;
  
      /**
       * The metadata image to extract topic id to names map.
       * This is initialised when the {@link GroupCoordinator#onNewMetadataImage(MetadataImage, MetadataDelta)} is called
       */
      private MetadataImage metadataImage = null;
  
      /**
       *
       * @param logContext                The log context.
       * @param config                    The group coordinator config.
       * @param runtime                   The runtime.
       * @param groupCoordinatorMetrics   The group coordinator metrics.
       * @param groupConfigManager        The group config manager.
       * @param persister                 The persister.
       * @param timer                     The timer.
       */
      GroupCoordinatorService(
          LogContext logContext,
          GroupCoordinatorConfig config,
          CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime,
          GroupCoordinatorMetrics groupCoordinatorMetrics,
          GroupConfigManager groupConfigManager,
          Persister persister,
          Timer timer
      ) {
          this.log = logContext.logger(GroupCoordinatorService.class);
          this.config = config;
          this.runtime = runtime;
          this.groupCoordinatorMetrics = groupCoordinatorMetrics;
          this.groupConfigManager = groupConfigManager;
          this.persister = persister;
          this.timer = timer;
          this.consumerGroupAssignors = config
              .consumerGroupAssignors()
              .stream()
              .map(ConsumerGroupPartitionAssignor::name)
              .collect(Collectors.toSet());
      }
  
      /**
       * Throws CoordinatorNotAvailableException if the not active.
       */
      private void throwIfNotActive() {
          if (!isActive.get()) {
              throw Errors.COORDINATOR_NOT_AVAILABLE.exception();
          }
      }
  
      /**
       * @return The topic partition for the given group.
       */
      private TopicPartition topicPartitionFor(
          String groupId
      ) {
          return new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, partitionFor(groupId));
      }
  
      /**
       * See {@link GroupCoordinator#partitionFor(String)}
       */
      @Override
      public int partitionFor(
          String groupId
      ) {
          throwIfNotActive();
          return Utils.abs(groupId.hashCode()) % numPartitions;
      }
  
      /**
       * Validates the request.
       *
       * @param request The request to validate.
       * @param apiVersion The version of ConsumerGroupHeartbeat RPC
       * @throws InvalidRequestException if the request is not valid.
       * @throws UnsupportedAssignorException if the assignor is not supported.
       */
      private void throwIfConsumerGroupHeartbeatRequestIsInvalid(
          ConsumerGroupHeartbeatRequestData request,
          int apiVersion
      ) throws InvalidRequestException, UnsupportedAssignorException {
          if (apiVersion >= CONSUMER_GENERATED_MEMBER_ID_REQUIRED_VERSION ||
              request.memberEpoch() > 0 ||
              request.memberEpoch() == LEAVE_GROUP_MEMBER_EPOCH
          ) {
              throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
          }
  
          throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
          throwIfEmptyString(request.instanceId(), "InstanceId can't be empty.");
          throwIfEmptyString(request.rackId(), "RackId can't be empty.");
  
          if (request.memberEpoch() == 0) {
              if (request.rebalanceTimeoutMs() == -1) {
                  throw new InvalidRequestException("RebalanceTimeoutMs must be provided in first request.");
              }
              if (request.topicPartitions() == null || !request.topicPartitions().isEmpty()) {
                  throw new InvalidRequestException("TopicPartitions must be empty when (re-)joining.");
              }
              // We accept members joining with an empty list of names or an empty regex. It basically
              // means that they are not subscribed to any topics, but they are part of the group.
              if (request.subscribedTopicNames() == null && request.subscribedTopicRegex() == null) {
                  throw new InvalidRequestException("Either SubscribedTopicNames or SubscribedTopicRegex must" +
                      " be non-null when (re-)joining.");
              }
          } else if (request.memberEpoch() == LEAVE_GROUP_STATIC_MEMBER_EPOCH) {
              throwIfNull(request.instanceId(), "InstanceId can't be null.");
          } else if (request.memberEpoch() < LEAVE_GROUP_STATIC_MEMBER_EPOCH) {
              throw new InvalidRequestException("MemberEpoch is invalid.");
          }
  
          if (request.serverAssignor() != null && !consumerGroupAssignors.contains(request.serverAssignor())) {
              throw new UnsupportedAssignorException("ServerAssignor " + request.serverAssignor()
                  + " is not supported. Supported assignors: " + String.join(", ", consumerGroupAssignors)
                  + ".");
          }
      }
  
      /**
       * See {@link GroupCoordinator#consumerGroupHeartbeat(AuthorizableRequestContext, ConsumerGroupHeartbeatRequestData)}.
       */
      @Override
      public CompletableFuture<ConsumerGroupHeartbeatResponseData> consumerGroupHeartbeat(
          AuthorizableRequestContext context,
          ConsumerGroupHeartbeatRequestData request
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(new ConsumerGroupHeartbeatResponseData()
                  .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
              );
          }
  
          try {
              throwIfConsumerGroupHeartbeatRequestIsInvalid(request, context.requestVersion());
          } catch (Throwable ex) {
              ApiError apiError = ApiError.fromThrowable(ex);
              return CompletableFuture.completedFuture(new ConsumerGroupHeartbeatResponseData()
                  .setErrorCode(apiError.error().code())
                  .setErrorMessage(apiError.message())
              );
          }
  
          return runtime.scheduleWriteOperation(
              "consumer-group-heartbeat",
              topicPartitionFor(request.groupId()),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.consumerGroupHeartbeat(context, request)
          ).exceptionally(exception -> handleOperationException(
              "consumer-group-heartbeat",
              request,
              exception,
              (error, message) -> new ConsumerGroupHeartbeatResponseData()
                  .setErrorCode(error.code())
                  .setErrorMessage(message),
              log
          ));
      }
  
      private static void throwIfInvalidTopology(
          StreamsGroupHeartbeatRequestData.Topology topology
      ) throws StreamsInvalidTopologyException {
          for (StreamsGroupHeartbeatRequestData.Subtopology subtopology: topology.subtopologies()) {
              for (StreamsGroupHeartbeatRequestData.TopicInfo topicInfo: subtopology.stateChangelogTopics()) {
                  if (topicInfo.partitions() != 0) {
                      throw new StreamsInvalidTopologyException(String.format(
                          "Changelog topic %s must have an undefined partition count, but it is set to %d.",
                          topicInfo.name(), topicInfo.partitions()
                      ));
                  }
              }
          }
      }
  
      /**
       * Validates the request.
       *
       * @param request The request to validate.
       * @throws InvalidRequestException if the request is not valid.
       * @throws UnsupportedAssignorException if the assignor is not supported.
       */
      private static void throwIfStreamsGroupHeartbeatRequestIsInvalid(
          StreamsGroupHeartbeatRequestData request
      ) throws InvalidRequestException {
          throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
          throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
          throwIfEmptyString(request.instanceId(), "InstanceId can't be empty.");
          throwIfEmptyString(request.rackId(), "RackId can't be empty.");
  
          if (request.memberEpoch() == 0) {
              if (request.rebalanceTimeoutMs() == -1) {
                  throw new InvalidRequestException("RebalanceTimeoutMs must be provided in first request.");
              }
              throwIfNotEmptyCollection(request.activeTasks(), "ActiveTasks must be empty when (re-)joining.");
              throwIfNotEmptyCollection(request.standbyTasks(), "StandbyTasks must be empty when (re-)joining.");
              throwIfNotEmptyCollection(request.warmupTasks(), "WarmupTasks must be empty when (re-)joining.");
              throwIfNull(request.topology(), "Topology must be non-null when (re-)joining.");
              if (request.topology() != null) {
                  throwIfInvalidTopology(request.topology());
              }
          } else if (request.memberEpoch() == LEAVE_GROUP_STATIC_MEMBER_EPOCH) {
              throwIfNull(request.instanceId(), "InstanceId can't be null.");
          } else if (request.memberEpoch() < LEAVE_GROUP_STATIC_MEMBER_EPOCH) {
              throw new InvalidRequestException(String.format("MemberEpoch is %d, but must be greater than or equal to -2.",
                  request.memberEpoch()));
          }
  
          if (request.activeTasks() != null || request.standbyTasks() != null || request.warmupTasks() != null) {
              throwIfNull(request.activeTasks(), "If one task-type is non-null, all must be non-null.");
              throwIfNull(request.standbyTasks(), "If one task-type is non-null, all must be non-null.");
              throwIfNull(request.warmupTasks(), "If one task-type is non-null, all must be non-null.");
          }
  
          if (request.memberEpoch() != 0) {
              throwIfNotNull(request.topology(), "Topology can only be provided when (re-)joining.");
          }
      }
  
      /**
       * See
       * {@link GroupCoordinator#streamsGroupHeartbeat(AuthorizableRequestContext, StreamsGroupHeartbeatRequestData)}.
       */
      @Override
      public CompletableFuture<StreamsGroupHeartbeatResult> streamsGroupHeartbeat(
          AuthorizableRequestContext context,
          StreamsGroupHeartbeatRequestData request
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(
                  new StreamsGroupHeartbeatResult(
                      new StreamsGroupHeartbeatResponseData().setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code()),
                      Map.of()
                  )
              );
          }
  
          try {
              throwIfStreamsGroupHeartbeatRequestIsInvalid(request);
          } catch (Throwable ex) {
              ApiError apiError = ApiError.fromThrowable(ex);
              return CompletableFuture.completedFuture(
                  new StreamsGroupHeartbeatResult(
                      new StreamsGroupHeartbeatResponseData()
                          .setErrorCode(apiError.error().code())
                          .setErrorMessage(apiError.message()),
                      Map.of()
                  )
              );
          }
  
          return runtime.scheduleWriteOperation(
              "streams-group-heartbeat",
              topicPartitionFor(request.groupId()),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.streamsGroupHeartbeat(context, request)
          ).exceptionally(exception -> handleOperationException(
              "streams-group-heartbeat",
              request,
              exception,
              (error, message) ->
                  new StreamsGroupHeartbeatResult(
                      new StreamsGroupHeartbeatResponseData()
                          .setErrorCode(error.code())
                          .setErrorMessage(message),
                      Map.of()
                  ),
              log
          ));
      }
  
      /**
       * Validates the ShareGroupHeartbeat request.
       *
       * @param request The request to validate.
       * @throws InvalidRequestException if the request is not valid.
       */
      private static void throwIfShareGroupHeartbeatRequestIsInvalid(
          ShareGroupHeartbeatRequestData request
      ) throws InvalidRequestException {
          throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
          throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
          throwIfEmptyString(request.rackId(), "RackId can't be empty.");
  
          if (request.memberEpoch() == 0) {
              if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
                  throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
              }
          } else if (request.memberEpoch() < ShareGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH) {
              throw new InvalidRequestException("MemberEpoch is invalid.");
          }
      }
  
      /**
       * See {@link GroupCoordinator#shareGroupHeartbeat(AuthorizableRequestContext, ShareGroupHeartbeatRequestData)}.
       */
      @Override
      public CompletableFuture<ShareGroupHeartbeatResponseData> shareGroupHeartbeat(
          AuthorizableRequestContext context,
          ShareGroupHeartbeatRequestData request
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(new ShareGroupHeartbeatResponseData()
                  .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
              );
          }
  
          try {
              throwIfShareGroupHeartbeatRequestIsInvalid(request);
          } catch (Throwable ex) {
              ApiError apiError = ApiError.fromThrowable(ex);
              return CompletableFuture.completedFuture(new ShareGroupHeartbeatResponseData()
                  .setErrorCode(apiError.error().code())
                  .setErrorMessage(apiError.message())
              );
          }
  
          return runtime.scheduleWriteOperation(
              "share-group-heartbeat",
              topicPartitionFor(request.groupId()),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.shareGroupHeartbeat(context, request)
          ).thenCompose(result -> {
              if (result.getValue().isPresent()) {
                  // Adding to timer makes this call async with respect to the heartbeat.
                  timer.add(new TimerTask(0L) {
                      @Override
                      public void run() {
                          persisterInitialize(result.getValue().get(), result.getKey())
                              .whenComplete((__, exp) -> {
                                  if (exp != null) {
                                      log.error("Persister initialization failed", exp);
                                  }
                              });
                      }
                  });
              }
              return CompletableFuture.completedFuture(result.getKey());
          }).exceptionally(exception -> handleOperationException(
              "share-group-heartbeat",
              request,
              exception,
              (error, message) -> new ShareGroupHeartbeatResponseData()
                  .setErrorCode(error.code())
                  .setErrorMessage(message),
              log
          ));
      }
  
      // Visibility for testing
      CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
          InitializeShareGroupStateParameters request,
          AlterShareGroupOffsetsResponseData response
      ) {
          return persister.initializeState(request)
              .handle((result, exp) -> {
                  if (exp == null) {
                      if (result.errorCounts().isEmpty()) {
                          handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
                          return response;
                      } else {
                          //TODO build new AlterShareGroupOffsetsResponseData for error response
                          return response;
                      }
                  } else {
                      return buildErrorResponse(request, response, exp);
                  }
  
              });
      }
  
      private AlterShareGroupOffsetsResponseData buildErrorResponse(InitializeShareGroupStateParameters request, AlterShareGroupOffsetsResponseData response, Throwable exp) {
          // build new AlterShareGroupOffsetsResponseData for error response
          AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
          GroupTopicPartitionData<PartitionStateData> gtp = request.groupTopicPartitionData();
          log.error("Unable to initialize share group state for {}, {} while altering share group offsets", gtp.groupId(), gtp.topicsData(), exp);
          Errors error = Errors.forException(exp);
          data.setErrorCode(error.code())
              .setErrorMessage(error.message())
              .setResponses(response.responses());
          data.setResponses(
              new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
                  .map(topic -> {
                      AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
                          .setTopicName(topic.topicName());
                      topic.partitions().forEach(partition -> {
                          AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition partitionData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition()
                              .setPartitionIndex(partition.partitionIndex())
                              .setErrorCode(error.code())
                              .setErrorMessage(error.message());
                          topicData.partitions().add(partitionData);
                      });
                      return topicData;
                  })
                  .iterator()));
          // don't uninitialized share group state here, as we regard this alter share group offsets request failed.
          return data;
      }
  
      // Visibility for testing
      CompletableFuture<ShareGroupHeartbeatResponseData> persisterInitialize(
          InitializeShareGroupStateParameters request,
          ShareGroupHeartbeatResponseData defaultResponse
      ) {
          return persister.initializeState(request)
              .handle((response, exp) -> {
                  if (exp == null) {
                      return handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), response, defaultResponse);
                  }
                  GroupTopicPartitionData<PartitionStateData> gtp = request.groupTopicPartitionData();
                  log.error("Unable to initialize share group state {}, {}", gtp.groupId(), gtp.topicsData(), exp);
                  Errors error = Errors.forException(exp);
                  Map<Uuid, Set<Integer>> topicPartitionMap = new HashMap<>();
                  gtp.topicsData().forEach(topicData -> topicPartitionMap.computeIfAbsent(topicData.topicId(), k -> new HashSet<>())
                      .addAll(topicData.partitions().stream().map(PartitionStateData::partition).collect(Collectors.toSet())));
                  return uninitializeShareGroupState(error, gtp.groupId(), topicPartitionMap);
              })
              .thenCompose(resp -> resp);
      }
  
      private CompletableFuture<ShareGroupHeartbeatResponseData> handlePersisterInitializeResponse(
          String groupId,
          InitializeShareGroupStateResult persisterInitializeResult,
          ShareGroupHeartbeatResponseData defaultResponse
      ) {
          Errors persisterError = Errors.NONE;
          for (TopicData<PartitionErrorData> topicData : persisterInitializeResult.topicsData()) {
              Optional<PartitionErrorData> errData = topicData.partitions().stream().filter(partition -> partition.errorCode() != Errors.NONE.code()).findAny();
              if (errData.isPresent()) {
                  persisterError = Errors.forCode(errData.get().errorCode());
                  break;
              }
          }
  
          Map<Uuid, Set<Integer>> topicPartitionMap = new HashMap<>();
          for (TopicData<PartitionErrorData> topicData : persisterInitializeResult.topicsData()) {
              topicPartitionMap.put(
                  topicData.topicId(),
                  topicData.partitions().stream().map(PartitionErrorData::partition).collect(Collectors.toSet())
              );
          }
  
          if (persisterError.code() == Errors.NONE.code()) {
              if (topicPartitionMap.isEmpty()) {
                  return CompletableFuture.completedFuture(defaultResponse);
              }
              return performShareGroupStateMetadataInitialize(groupId, topicPartitionMap, defaultResponse);
          }
          log.error("Received error while calling initialize state for {} on persister {}.", groupId, persisterError.code());
          return uninitializeShareGroupState(persisterError, groupId, topicPartitionMap);
      }
  
      private CompletableFuture<ShareGroupHeartbeatResponseData> uninitializeShareGroupState(
          Errors error,
          String groupId,
          Map<Uuid, Set<Integer>> topicPartitionMap
      ) {
          return runtime.scheduleWriteOperation(
              "uninitialize-share-group-state",
              topicPartitionFor(groupId),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.uninitializeShareGroupState(groupId, topicPartitionMap)
          ).thenApply(__ -> new ShareGroupHeartbeatResponseData()
              .setErrorCode(error.code())
              .setErrorMessage(error.message())
          ).exceptionally(exception -> {
              log.error("Unable to cleanup topic partitions from share group state metadata", exception);
              Errors err = Errors.forException(new IllegalStateException("Unable to cleanup topic partitions from share group state metadata", exception));
              return new ShareGroupHeartbeatResponseData()
                  .setErrorCode(err.code())
                  .setErrorMessage(err.message());
          });
      }
  
      private CompletableFuture<ShareGroupHeartbeatResponseData> performShareGroupStateMetadataInitialize(
          String groupId,
          Map<Uuid, Set<Integer>> topicPartitionMap,
          ShareGroupHeartbeatResponseData defaultResponse
      ) {
          return runtime.scheduleWriteOperation(
              "initialize-share-group-state",
              topicPartitionFor(groupId),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.initializeShareGroupState(groupId, topicPartitionMap)
          ).handle((__, exp) -> {
              if (exp == null) {
                  return CompletableFuture.completedFuture(defaultResponse);
              }
              log.error("Unable to initialize share group state partition metadata for {}.", groupId, exp);
              Errors error = Errors.forException(exp);
              return uninitializeShareGroupState(error, groupId, topicPartitionMap);
          }).thenCompose(resp -> resp);
      }
  
      /**
       * See {@link GroupCoordinator#joinGroup(AuthorizableRequestContext, JoinGroupRequestData, BufferSupplier)}.
       */
      @Override
      public CompletableFuture<JoinGroupResponseData> joinGroup(
          AuthorizableRequestContext context,
          JoinGroupRequestData request,
          BufferSupplier bufferSupplier
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(new JoinGroupResponseData()
                  .setMemberId(request.memberId())
                  .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
              );
          }
  
          if (!isGroupIdNotEmpty(request.groupId())) {
              return CompletableFuture.completedFuture(new JoinGroupResponseData()
                  .setMemberId(request.memberId())
                  .setErrorCode(Errors.INVALID_GROUP_ID.code())
              );
          }
  
          if (request.sessionTimeoutMs() < config.classicGroupMinSessionTimeoutMs() ||
              request.sessionTimeoutMs() > config.classicGroupMaxSessionTimeoutMs()) {
              return CompletableFuture.completedFuture(new JoinGroupResponseData()
                  .setMemberId(request.memberId())
                  .setErrorCode(Errors.INVALID_SESSION_TIMEOUT.code())
              );
          }
  
          CompletableFuture<JoinGroupResponseData> responseFuture = new CompletableFuture<>();
  
          runtime.scheduleWriteOperation(
              "classic-group-join",
              topicPartitionFor(request.groupId()),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.classicGroupJoin(context, request, responseFuture)
          ).exceptionally(exception -> {
              if (!responseFuture.isDone()) {
                  responseFuture.complete(handleOperationException(
                      "classic-group-join",
                      request,
                      exception,
                      (error, __) -> new JoinGroupResponseData().setErrorCode(error.code()),
                      log
                  ));
              }
              return null;
          });
  
          return responseFuture;
      }
  
      /**
       * See {@link GroupCoordinator#syncGroup(AuthorizableRequestContext, SyncGroupRequestData, BufferSupplier)}.
       */
      @Override
      public CompletableFuture<SyncGroupResponseData> syncGroup(
          AuthorizableRequestContext context,
          SyncGroupRequestData request,
          BufferSupplier bufferSupplier
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(new SyncGroupResponseData()
                  .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
              );
          }
  
          if (!isGroupIdNotEmpty(request.groupId())) {
              return CompletableFuture.completedFuture(new SyncGroupResponseData()
                  .setErrorCode(Errors.INVALID_GROUP_ID.code())
              );
          }
  
          CompletableFuture<SyncGroupResponseData> responseFuture = new CompletableFuture<>();
  
          runtime.scheduleWriteOperation(
              "classic-group-sync",
              topicPartitionFor(request.groupId()),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.classicGroupSync(context, request, responseFuture)
          ).exceptionally(exception -> {
              if (!responseFuture.isDone()) {
                  responseFuture.complete(handleOperationException(
                      "classic-group-sync",
                      request,
                      exception,
                      (error, __) -> new SyncGroupResponseData().setErrorCode(error.code()),
                      log
                  ));
              }
              return null;
          });
  
          return responseFuture;
      }
  
      /**
       * See {@link GroupCoordinator#heartbeat(AuthorizableRequestContext, HeartbeatRequestData)}.
       */
      @Override
      public CompletableFuture<HeartbeatResponseData> heartbeat(
          AuthorizableRequestContext context,
          HeartbeatRequestData request
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(new HeartbeatResponseData()
                  .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
              );
          }
  
          if (!isGroupIdNotEmpty(request.groupId())) {
              return CompletableFuture.completedFuture(new HeartbeatResponseData()
                  .setErrorCode(Errors.INVALID_GROUP_ID.code())
              );
          }
  
          return runtime.scheduleWriteOperation(
              "classic-group-heartbeat",
              topicPartitionFor(request.groupId()),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.classicGroupHeartbeat(context, request)
          ).exceptionally(exception -> handleOperationException(
              "classic-group-heartbeat",
              request,
              exception,
              (error, __) -> {
                  if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
                      // The group is still loading, so blindly respond
                      return new HeartbeatResponseData()
                          .setErrorCode(Errors.NONE.code());
                  } else {
                      return new HeartbeatResponseData()
                          .setErrorCode(error.code());
                  }
              },
              log
          ));
      }
  
      /**
       * See {@link GroupCoordinator#leaveGroup(AuthorizableRequestContext, LeaveGroupRequestData)}.
       */
      @Override
      public CompletableFuture<LeaveGroupResponseData> leaveGroup(
          AuthorizableRequestContext context,
          LeaveGroupRequestData request
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(new LeaveGroupResponseData()
                  .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
              );
          }
  
          if (!isGroupIdNotEmpty(request.groupId())) {
              return CompletableFuture.completedFuture(new LeaveGroupResponseData()
                  .setErrorCode(Errors.INVALID_GROUP_ID.code())
              );
          }
  
          return runtime.scheduleWriteOperation(
              "classic-group-leave",
              topicPartitionFor(request.groupId()),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.classicGroupLeave(context, request)
          ).exceptionally(exception -> handleOperationException(
              "classic-group-leave",
              request,
              exception,
              (error, __) -> {
                  if (error == Errors.UNKNOWN_MEMBER_ID) {
                      // Group was not found.
                      List<LeaveGroupResponseData.MemberResponse> memberResponses = request.members().stream()
                           .map(member -> new LeaveGroupResponseData.MemberResponse()
                               .setMemberId(member.memberId())
                               .setGroupInstanceId(member.groupInstanceId())
                               .setErrorCode(Errors.UNKNOWN_MEMBER_ID.code()))
                           .toList();
                      return new LeaveGroupResponseData()
                          .setMembers(memberResponses);
                  } else {
                      return new LeaveGroupResponseData()
                          .setErrorCode(error.code());
                  }
              },
              log
          ));
      }
  
      /**
       * See {@link GroupCoordinator#listGroups(AuthorizableRequestContext, ListGroupsRequestData)}.
       */
      @Override
      public CompletableFuture<ListGroupsResponseData> listGroups(
          AuthorizableRequestContext context,
          ListGroupsRequestData request
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(new ListGroupsResponseData()
                  .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
              );
          }
  
          final List<CompletableFuture<List<ListGroupsResponseData.ListedGroup>>> futures = FutureUtils.mapExceptionally(
              runtime.scheduleReadAllOperation(
                  "list-groups",
                  (coordinator, lastCommittedOffset) -> coordinator.listGroups(
                      request.statesFilter(),
                      request.typesFilter(),
                      lastCommittedOffset
                  )
              ),
              exception -> {
                  exception = Errors.maybeUnwrapException(exception);
                  if (exception instanceof NotCoordinatorException) {
                      return List.of();
                  } else {
                      throw new CompletionException(exception);
                  }
              }
          );
  
          return FutureUtils
              .combineFutures(futures, ArrayList::new, List::addAll)
              .thenApply(groups -> new ListGroupsResponseData().setGroups(groups))
              .exceptionally(exception -> handleOperationException(
                  "list-groups",
                  request,
                  exception,
                  (error, __) -> new ListGroupsResponseData().setErrorCode(error.code()),
                  log
              ));
      }
  
      /**
       * See {@link GroupCoordinator#consumerGroupDescribe(AuthorizableRequestContext, List)}.
       */
      @Override
      public CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>> consumerGroupDescribe(
          AuthorizableRequestContext context,
          List<String> groupIds
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(ConsumerGroupDescribeRequest.getErrorDescribedGroupList(
                  groupIds,
                  Errors.COORDINATOR_NOT_AVAILABLE
              ));
          }
  
          final List<CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>>> futures =
              new ArrayList<>(groupIds.size());
          final Map<TopicPartition, List<String>> groupsByTopicPartition = new HashMap<>();
          groupIds.forEach(groupId -> {
              if (isGroupIdNotEmpty(groupId)) {
                  groupsByTopicPartition
                      .computeIfAbsent(topicPartitionFor(groupId), __ -> new ArrayList<>())
                      .add(groupId);
              } else {
                  futures.add(CompletableFuture.completedFuture(List.of(
                      new ConsumerGroupDescribeResponseData.DescribedGroup()
                          .setGroupId(null)
                          .setErrorCode(Errors.INVALID_GROUP_ID.code())
                  )));
              }
          });
  
          groupsByTopicPartition.forEach((topicPartition, groupList) -> {
              CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>> future =
                  runtime.scheduleReadOperation(
                      "consumer-group-describe",
                      topicPartition,
                      (coordinator, lastCommittedOffset) -> coordinator.consumerGroupDescribe(groupList, lastCommittedOffset)
                  ).exceptionally(exception -> handleOperationException(
                      "consumer-group-describe",
                      groupList,
                      exception,
                      (error, __) -> ConsumerGroupDescribeRequest.getErrorDescribedGroupList(groupList, error),
                      log
                  ));
  
              futures.add(future);
          });
  
          return FutureUtils.combineFutures(futures, ArrayList::new, List::addAll);
      }
  
      /**
       * See {@link GroupCoordinator#streamsGroupDescribe(AuthorizableRequestContext, List)}.
       */
      @Override
      public CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>> streamsGroupDescribe(
          AuthorizableRequestContext context,
          List<String> groupIds
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(StreamsGroupDescribeRequest.getErrorDescribedGroupList(
                  groupIds,
                  Errors.COORDINATOR_NOT_AVAILABLE
              ));
          }
  
          final List<CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>>> futures =
              new ArrayList<>(groupIds.size());
          final Map<TopicPartition, List<String>> groupsByTopicPartition = new HashMap<>();
          groupIds.forEach(groupId -> {
              if (isGroupIdNotEmpty(groupId)) {
                  groupsByTopicPartition
                      .computeIfAbsent(topicPartitionFor(groupId), __ -> new ArrayList<>())
                      .add(groupId);
              } else {
                  futures.add(CompletableFuture.completedFuture(List.of(
                      new StreamsGroupDescribeResponseData.DescribedGroup()
                          .setGroupId(null)
                          .setErrorCode(Errors.INVALID_GROUP_ID.code())
                  )));
              }
          });
  
          groupsByTopicPartition.forEach((topicPartition, groupList) -> {
              CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>> future =
                  runtime.scheduleReadOperation(
                      "streams-group-describe",
                      topicPartition,
                      (coordinator, lastCommittedOffset) -> coordinator.streamsGroupDescribe(groupList, lastCommittedOffset)
                  ).exceptionally(exception -> handleOperationException(
                      "streams-group-describe",
                      groupList,
                      exception,
                      (error, __) -> StreamsGroupDescribeRequest.getErrorDescribedGroupList(groupList, error),
                      log
                  ));
  
              futures.add(future);
          });
  
          return FutureUtils.combineFutures(futures, ArrayList::new, List::addAll);
      }
      
      /**
       * See {@link GroupCoordinator#shareGroupDescribe(AuthorizableRequestContext, List)}.
       */
      @Override
      public CompletableFuture<List<DescribedGroup>> shareGroupDescribe(
          AuthorizableRequestContext context,
          List<String> groupIds
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(ShareGroupDescribeRequest.getErrorDescribedGroupList(
                  groupIds,
                  Errors.COORDINATOR_NOT_AVAILABLE
              ));
          }
  
          final List<CompletableFuture<List<ShareGroupDescribeResponseData.DescribedGroup>>> futures =
              new ArrayList<>(groupIds.size());
          final Map<TopicPartition, List<String>> groupsByTopicPartition = new HashMap<>();
          groupIds.forEach(groupId -> {
              if (isGroupIdNotEmpty(groupId)) {
                  groupsByTopicPartition
                      .computeIfAbsent(topicPartitionFor(groupId), __ -> new ArrayList<>())
                      .add(groupId);
              } else {
                  futures.add(CompletableFuture.completedFuture(List.of(
                      new ShareGroupDescribeResponseData.DescribedGroup()
                          .setGroupId(null)
                          .setErrorCode(Errors.INVALID_GROUP_ID.code())
                  )));
              }
          });
  
          groupsByTopicPartition.forEach((topicPartition, groupList) -> {
              CompletableFuture<List<ShareGroupDescribeResponseData.DescribedGroup>> future =
                  runtime.scheduleReadOperation(
                      "share-group-describe",
                      topicPartition,
                      (coordinator, lastCommittedOffset) -> coordinator.shareGroupDescribe(groupList, lastCommittedOffset)
                  ).exceptionally(exception -> handleOperationException(
                      "share-group-describe",
                      groupList,
                      exception,
                      (error, __) -> ShareGroupDescribeRequest.getErrorDescribedGroupList(groupList, error),
                      log
                  ));
  
              futures.add(future);
          });
  
          return FutureUtils.combineFutures(futures, ArrayList::new, List::addAll);
      }
  
      /**
       * See {@link GroupCoordinator#alterShareGroupOffsets(AuthorizableRequestContext, String, AlterShareGroupOffsetsRequestData)}.
       */
      @Override
      public CompletableFuture<AlterShareGroupOffsetsResponseData> alterShareGroupOffsets(AuthorizableRequestContext context, String groupId, AlterShareGroupOffsetsRequestData request) {
          if (!isActive.get() || metadataImage == null) {
              return CompletableFuture.completedFuture(AlterShareGroupOffsetsRequest.getErrorResponse(Errors.COORDINATOR_NOT_AVAILABLE));
          }
          
          if (groupId == null || groupId.isEmpty()) {
              return CompletableFuture.completedFuture(AlterShareGroupOffsetsRequest.getErrorResponse(Errors.INVALID_GROUP_ID));
          }
  
          if (request.topics() == null || request.topics().isEmpty()) {
              return CompletableFuture.completedFuture(new AlterShareGroupOffsetsResponseData());
          }
  
          return runtime.scheduleWriteOperation(
              "share-group-offsets-alter",
              topicPartitionFor(groupId),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.alterShareGroupOffsets(groupId, request)
          ).thenCompose(result ->
              persisterInitialize(result.getValue(), result.getKey())
          ).exceptionally(exception -> handleOperationException(
              "share-group-offsets-alter",
              request,
              exception,
              (error, message) -> AlterShareGroupOffsetsRequest.getErrorResponse(error),
              log
          ));
      }
  
      /**
       * See {@link GroupCoordinator#describeGroups(AuthorizableRequestContext, List)}.
       */
      @Override
      public CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>> describeGroups(
          AuthorizableRequestContext context,
          List<String> groupIds
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(DescribeGroupsRequest.getErrorDescribedGroupList(
                  groupIds,
                  Errors.COORDINATOR_NOT_AVAILABLE
              ));
          }
  
          final List<CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>>> futures =
              new ArrayList<>(groupIds.size());
          final Map<TopicPartition, List<String>> groupsByTopicPartition = new HashMap<>();
          groupIds.forEach(groupId -> {
              // For backwards compatibility, we support DescribeGroups for the empty group id.
              if (groupId == null) {
                  futures.add(CompletableFuture.completedFuture(List.of(
                      new DescribeGroupsResponseData.DescribedGroup()
                          .setGroupId(null)
                          .setErrorCode(Errors.INVALID_GROUP_ID.code())
                  )));
              } else {
                  final TopicPartition topicPartition = topicPartitionFor(groupId);
                  groupsByTopicPartition
                      .computeIfAbsent(topicPartition, __ -> new ArrayList<>())
                      .add(groupId);
              }
          });
  
          groupsByTopicPartition.forEach((topicPartition, groupList) -> {
              CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>> future =
                  runtime.scheduleReadOperation(
                      "describe-groups",
                      topicPartition,
                      (coordinator, lastCommittedOffset) -> coordinator.describeGroups(context, groupList, lastCommittedOffset)
                  ).exceptionally(exception -> handleOperationException(
                      "describe-groups",
                      groupList,
                      exception,
                      (error, __) -> DescribeGroupsRequest.getErrorDescribedGroupList(groupList, error),
                      log
                  ));
  
              futures.add(future);
          });
  
          return FutureUtils.combineFutures(futures, ArrayList::new, List::addAll);
      }
  
      /**
       * See {@link GroupCoordinator#deleteGroups(AuthorizableRequestContext, List, BufferSupplier)}.
       */
      @Override
      public CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> deleteGroups(
          AuthorizableRequestContext context,
          List<String> groupIds,
          BufferSupplier bufferSupplier
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(DeleteGroupsRequest.getErrorResultCollection(
                  groupIds,
                  Errors.COORDINATOR_NOT_AVAILABLE
              ));
          }
  
          final List<CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection>> futures =
              new ArrayList<>(groupIds.size());
  
          final Map<TopicPartition, List<String>> groupsByTopicPartition = new HashMap<>();
          groupIds.forEach(groupId -> {
              // For backwards compatibility, we support DeleteGroups for the empty group id.
              if (groupId == null) {
                  futures.add(CompletableFuture.completedFuture(DeleteGroupsRequest.getErrorResultCollection(
                      Collections.singletonList(null),
                      Errors.INVALID_GROUP_ID
                  )));
              } else {
                  final TopicPartition topicPartition = topicPartitionFor(groupId);
                  groupsByTopicPartition
                      .computeIfAbsent(topicPartition, __ -> new ArrayList<>())
                      .add(groupId);
              }
          });
  
          groupsByTopicPartition.forEach((topicPartition, groupList) -> {
              // Since the specific group types are not known, group deletion operations are chained.
              // The sequence of these deletions is important: the initial share group deletion should
              // not error if the group ID isn't found, whereas the subsequent consumer group deletion
              // (the final operation) must return an error if its corresponding group ID is not found.
              CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> future = deleteShareGroups(topicPartition, groupList).thenCompose(groupErrMap -> {
                  DeleteGroupsResponseData.DeletableGroupResultCollection deletableGroupResults = new DeleteGroupsResponseData.DeletableGroupResultCollection();
                  List<String> retainedGroupIds = updateResponseAndGetNonErrorGroupList(groupErrMap, groupList, deletableGroupResults);
                  if (retainedGroupIds.isEmpty()) {
                      return CompletableFuture.completedFuture(deletableGroupResults);
                  }
  
                  return handleDeleteGroups(context, topicPartition, retainedGroupIds)
                      .whenComplete((resp, __) -> resp.forEach(result -> deletableGroupResults.add(result.duplicate())))
                      .thenApply(__ -> deletableGroupResults);
              });
              // deleteShareGroups has its own exceptionally block, so we don't need one here.
  
              // This future object has the following stages:
              // - First it invokes the share group delete flow where the shard sharePartitionDeleteRequests
              // method is invoked, and it returns request objects for each valid share group passed to it.
              // All initialized and initializing share partitions are moved to deleting.
              // - Then the requests are passed to the persister.deleteState method one at a time. The results
              // are collated as a Map of groupId -> persister errors
              // - The above map can be used to decide whether to invoke the group coordinator delete groups logic
              // - Share groups with failed persister delete are NOT CONSIDERED for group coordinator delete.
              // TLDR: DeleteShareGroups -> filter erroneous persister deletes -> general delete groups logic
              futures.add(future);
          });
  
          return FutureUtils.combineFutures(futures, DeleteGroupsResponseData.DeletableGroupResultCollection::new,
              // We don't use res.addAll(future.join()) because DeletableGroupResultCollection is an ImplicitLinkedHashMultiCollection,
              // which has requirements for adding elements (see ImplicitLinkedHashCollection.java#add).
              (accumulator, newResults) -> newResults.forEach(result -> accumulator.add(result.duplicate())));
      }
  
      /**
       * Processes input shareGroupErrMap by retaining only those which do not contain an error.
       * Also updates the result collection input arg with share groups containing errors.
       *
       * @param shareGroupErrMap      Map keyed on share groupId and value as the error (NONE for no error).
       * @param groupList             Entire list of groups (all types)
       * @param deletableGroupResults Collection of responses for delete groups request.
       * @return A list of all non-error groupIds
       */
      private List<String> updateResponseAndGetNonErrorGroupList(
          Map<String, Errors> shareGroupErrMap,
          List<String> groupList,
          DeleteGroupsResponseData.DeletableGroupResultCollection deletableGroupResults
      ) {
          List<String> errGroupIds = new ArrayList<>();
          shareGroupErrMap.forEach((groupId, error) -> {
              if (error.code() != Errors.NONE.code()) {
                  log.error("Error deleting share group {} due to error {}", groupId, error);
                  errGroupIds.add(groupId);
                  deletableGroupResults.add(
                      new DeleteGroupsResponseData.DeletableGroupResult()
                          .setGroupId(groupId)
                          .setErrorCode(error.code())
                  );
              }
          });
  
          Set<String> groupSet = new HashSet<>(groupList);
          // Remove all share group ids which have errored out
          // when deleting with persister.
          errGroupIds.forEach(groupSet::remove);
  
          // Let us invoke the standard procedure of any non-share
          // groups or successfully deleted share groups remaining.
          return groupSet.stream().toList();
      }
  
      private CompletableFuture<DeleteGroupsResponseData.DeletableGroupResultCollection> handleDeleteGroups(
          AuthorizableRequestContext context,
          TopicPartition topicPartition,
          List<String> groupIds
      ) {
          return runtime.scheduleWriteOperation(
              "delete-groups",
              topicPartition,
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.deleteGroups(context, groupIds)
          ).exceptionally(exception -> handleOperationException(
              "delete-groups",
              groupIds,
              exception,
              (error, __) -> DeleteGroupsRequest.getErrorResultCollection(groupIds, error),
              log
          ));
      }
  
      private CompletableFuture<Map<String, Errors>> deleteShareGroups(
          TopicPartition topicPartition,
          List<String> groupList
      ) {
          // topicPartition refers to internal topic __consumer_offsets.
          return runtime.scheduleWriteOperation(
              "delete-share-groups",
              topicPartition,
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.sharePartitionDeleteRequests(groupList)
          ).thenCompose(
              this::performShareGroupsDeletion
          ).exceptionally(exception -> handleOperationException(
              "delete-share-groups",
              groupList,
              exception,
              (error, __) -> {
                  Map<String, Errors> errors = new HashMap<>();
                  groupList.forEach(group -> errors.put(group, error));
                  return errors;
              },
              log
          ));
      }
  
      private CompletableFuture<Map<String, Errors>> performShareGroupsDeletion(
          Map<String, Map.Entry<DeleteShareGroupStateParameters, Errors>> deleteRequests
      ) {
          List<CompletableFuture<Map.Entry<String, DeleteShareGroupStateResult>>> futures = new ArrayList<>(deleteRequests.size());
          Map<String, Errors> errorMap = new HashMap<>();
          deleteRequests.forEach((groupId, valPair) -> {
              if (valPair.getValue() == Errors.NONE) {
                  futures.add(deleteShareGroup(valPair.getKey()));
              } else {
                  errorMap.put(groupId, valPair.getValue());
              }
          });
  
          return persisterDeleteToGroupIdErrorMap(futures)
              .thenApply(respErrMap -> {
                  errorMap.putAll(respErrMap);
                  return errorMap;
              });
      }
  
      private CompletableFuture<Map.Entry<String, DeleteShareGroupStateResult>> deleteShareGroup(
          DeleteShareGroupStateParameters deleteRequest
      ) {
          String groupId = deleteRequest.groupTopicPartitionData().groupId();
          return persister.deleteState(deleteRequest)
              .thenCompose(result -> CompletableFuture.completedFuture(Map.entry(groupId, result)))
              .exceptionally(exception -> {
                  // In case the deleteState call fails,
                  // we should construct the appropriate response here
                  // so that the subsequent callbacks don't see runtime exceptions.
                  log.error("Unable to delete share group partition(s) - {} using request {}", groupId, deleteRequest, exception);
                  List<TopicData<PartitionErrorData>> respTopicData = deleteRequest.groupTopicPartitionData().topicsData().stream()
                      .map(reqTopicData -> new TopicData<>(
                          reqTopicData.topicId(),
                          reqTopicData.partitions().stream()
                              .map(reqPartData -> {
                                  Errors err = Errors.forException(exception);
                                  return PartitionFactory.newPartitionErrorData(reqPartData.partition(), err.code(), err.message());
                              })
                              .toList()
                      ))
                      .toList();
  
                  return Map.entry(groupId, new DeleteShareGroupStateResult.Builder()
                      .setTopicsData(respTopicData)
                      .build()
                  );
              });
      }
  
      private CompletableFuture<Map<String, Errors>> persisterDeleteToGroupIdErrorMap(
          List<CompletableFuture<Map.Entry<String, DeleteShareGroupStateResult>>> futures
      ) {
          return CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[]{})).thenCompose(v -> {
              Map<String, Errors> groupIds = new HashMap<>();
              for (CompletableFuture<Map.Entry<String, DeleteShareGroupStateResult>> future : futures) {
                  Map.Entry<String, DeleteShareGroupStateResult> entry = future.getNow(null);  // safe as within allOff
                  groupIds.putIfAbsent(entry.getKey(), Errors.NONE);
                  for (TopicData<PartitionErrorData> topicData : entry.getValue().topicsData()) {
                      Optional<PartitionErrorData> errItem = topicData.partitions().stream()
                          .filter(errData -> errData.errorCode() != Errors.NONE.code())
                          .findAny();
  
                      errItem.ifPresent(val -> {
                          log.error("Received error while deleting share group {} - {}", entry.getKey(), val);
                          groupIds.put(entry.getKey(), Errors.forCode(val.errorCode()));
                      });
                  }
              }
  
              return CompletableFuture.completedFuture(groupIds);
          });
      }
  
      /**
       * See {@link GroupCoordinator#fetchOffsets(AuthorizableRequestContext, OffsetFetchRequestData.OffsetFetchRequestGroup, boolean)}.
       */
      @Override
      public CompletableFuture<OffsetFetchResponseData.OffsetFetchResponseGroup> fetchOffsets(
          AuthorizableRequestContext context,
          OffsetFetchRequestData.OffsetFetchRequestGroup request,
          boolean requireStable
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(new OffsetFetchResponseData.OffsetFetchResponseGroup()
                  .setGroupId(request.groupId())
                  .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
              );
          }
  
          // For backwards compatibility, we support fetch commits for the empty group id.
          if (request.groupId() == null) {
              return CompletableFuture.completedFuture(new OffsetFetchResponseData.OffsetFetchResponseGroup()
                  .setGroupId(request.groupId())
                  .setErrorCode(Errors.INVALID_GROUP_ID.code())
              );
          }
  
          // The require stable flag when set tells the broker to hold on returning unstable
          // (or uncommitted) offsets. In the previous implementation of the group coordinator,
          // the UNSTABLE_OFFSET_COMMIT error is returned when unstable offsets are present. As
          // the new implementation relies on timeline data structures, the coordinator does not
          // really know whether offsets are stable or not so it is hard to return the same error.
          // Instead, we use a write operation when the flag is set to guarantee that the fetch
          // is based on all the available offsets and to ensure that the response waits until
          // the pending offsets are committed. Otherwise, we use a read operation.
          if (requireStable) {
              return runtime.scheduleWriteOperation(
                  "fetch-offsets",
                  topicPartitionFor(request.groupId()),
                  Duration.ofMillis(config.offsetCommitTimeoutMs()),
                  coordinator -> new CoordinatorResult<>(
                      List.of(),
                      coordinator.fetchOffsets(request, Long.MAX_VALUE)
                  )
              ).exceptionally(exception -> handleOffsetFetchException(
                  "fetch-offsets",
                  request,
                  exception
              ));
          } else {
              return runtime.scheduleReadOperation(
                  "fetch-offsets",
                  topicPartitionFor(request.groupId()),
                  (coordinator, offset) -> coordinator.fetchOffsets(request, offset)
              );
          }
      }
  
      /**
       * See {@link GroupCoordinator#fetchAllOffsets(AuthorizableRequestContext, OffsetFetchRequestData.OffsetFetchRequestGroup, boolean)}.
       */
      @Override
      public CompletableFuture<OffsetFetchResponseData.OffsetFetchResponseGroup> fetchAllOffsets(
          AuthorizableRequestContext context,
          OffsetFetchRequestData.OffsetFetchRequestGroup request,
          boolean requireStable
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(new OffsetFetchResponseData.OffsetFetchResponseGroup()
                  .setGroupId(request.groupId())
                  .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
              );
          }
  
          // For backwards compatibility, we support fetch commits for the empty group id.
          if (request.groupId() == null) {
              return CompletableFuture.completedFuture(new OffsetFetchResponseData.OffsetFetchResponseGroup()
                  .setGroupId(request.groupId())
                  .setErrorCode(Errors.INVALID_GROUP_ID.code())
              );
          }
  
          // The require stable flag when set tells the broker to hold on returning unstable
          // (or uncommitted) offsets. In the previous implementation of the group coordinator,
          // the UNSTABLE_OFFSET_COMMIT error is returned when unstable offsets are present. As
          // the new implementation relies on timeline data structures, the coordinator does not
          // really know whether offsets are stable or not so it is hard to return the same error.
          // Instead, we use a write operation when the flag is set to guarantee that the fetch
          // is based on all the available offsets and to ensure that the response waits until
          // the pending offsets are committed. Otherwise, we use a read operation.
          if (requireStable) {
              return runtime.scheduleWriteOperation(
                  "fetch-all-offsets",
                  topicPartitionFor(request.groupId()),
                  Duration.ofMillis(config.offsetCommitTimeoutMs()),
                  coordinator -> new CoordinatorResult<>(
                      List.of(),
                      coordinator.fetchAllOffsets(request, Long.MAX_VALUE)
                  )
              ).exceptionally(exception -> handleOffsetFetchException(
                  "fetch-all-offsets",
                  request,
                  exception
              ));
          } else {
              return runtime.scheduleReadOperation(
                  "fetch-all-offsets",
                  topicPartitionFor(request.groupId()),
                  (coordinator, offset) -> coordinator.fetchAllOffsets(request, offset)
              );
          }
      }
  
      /**
       * See {@link GroupCoordinator#describeShareGroupOffsets(AuthorizableRequestContext, DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup)}.
       */
      @Override
      public CompletableFuture<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup> describeShareGroupOffsets(
          AuthorizableRequestContext context,
          DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup requestData
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(
                  DescribeShareGroupOffsetsRequest.getErrorDescribedGroup(requestData.groupId(), Errors.COORDINATOR_NOT_AVAILABLE));
          }
  
          if (metadataImage == null) {
              return CompletableFuture.completedFuture(
                  DescribeShareGroupOffsetsRequest.getErrorDescribedGroup(requestData.groupId(), Errors.COORDINATOR_NOT_AVAILABLE));
          }
  
          Map<Uuid, String> requestTopicIdToNameMapping = new HashMap<>();
          List<ReadShareGroupStateSummaryRequestData.ReadStateSummaryData> readStateSummaryData = new ArrayList<>(requestData.topics().size());
          List<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseTopic> describeShareGroupOffsetsResponseTopicList = new ArrayList<>(requestData.topics().size());
          requestData.topics().forEach(topic -> {
              Uuid topicId = metadataImage.topics().topicNameToIdView().get(topic.topicName());
              if (topicId != null) {
                  requestTopicIdToNameMapping.put(topicId, topic.topicName());
                  readStateSummaryData.add(new ReadShareGroupStateSummaryRequestData.ReadStateSummaryData()
                      .setTopicId(topicId)
                      .setPartitions(
                          topic.partitions().stream().map(
                              partitionIndex -> new ReadShareGroupStateSummaryRequestData.PartitionData().setPartition(partitionIndex)
                          ).toList()
                      ));
              } else {
                  // If the topic does not exist, the start offset is returned as -1 (uninitialized offset).
                  // This is consistent with OffsetFetch for situations in which there is no offset information to fetch.
                  // It's treated as absence of data, rather than an error, unlike TOPIC_AUTHORIZATION_ERROR for example.
                  describeShareGroupOffsetsResponseTopicList.add(new DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseTopic()
                      .setTopicName(topic.topicName())
                      .setTopicId(Uuid.ZERO_UUID)
                      .setPartitions(topic.partitions().stream().map(
                          partition -> new DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponsePartition()
                              .setPartitionIndex(partition)
                              .setStartOffset(PartitionFactory.UNINITIALIZED_START_OFFSET)
                      ).toList()));
              }
          });
  
          // If the request for the persister is empty, just complete the operation right away.
          if (readStateSummaryData.isEmpty()) {
              return CompletableFuture.completedFuture(
                  new DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup()
                      .setGroupId(requestData.groupId())
                      .setTopics(describeShareGroupOffsetsResponseTopicList));
          }
  
          ReadShareGroupStateSummaryRequestData readSummaryRequestData = new ReadShareGroupStateSummaryRequestData()
              .setGroupId(requestData.groupId())
              .setTopics(readStateSummaryData);
  
          return readShareGroupStateSummary(readSummaryRequestData, requestTopicIdToNameMapping, describeShareGroupOffsetsResponseTopicList);
      }
  
      /**
       * See {@link GroupCoordinator#describeShareGroupAllOffsets(AuthorizableRequestContext, DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup)}.
       */
      @Override
      public CompletableFuture<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup> describeShareGroupAllOffsets(
          AuthorizableRequestContext context,
          DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup requestData
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(
                  DescribeShareGroupOffsetsRequest.getErrorDescribedGroup(requestData.groupId(), Errors.COORDINATOR_NOT_AVAILABLE));
          }
  
          if (metadataImage == null) {
              return CompletableFuture.completedFuture(
                  DescribeShareGroupOffsetsRequest.getErrorDescribedGroup(requestData.groupId(), Errors.COORDINATOR_NOT_AVAILABLE));
          }
  
          return runtime.scheduleReadOperation(
              "share-group-initialized-partitions",
              topicPartitionFor(requestData.groupId()),
              (coordinator, offset) -> coordinator.initializedShareGroupPartitions(requestData.groupId())
          ).thenCompose(topicPartitionMap -> {
              Map<Uuid, String> requestTopicIdToNameMapping = new HashMap<>();
              List<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseTopic> describeShareGroupOffsetsResponseTopicList = new ArrayList<>(topicPartitionMap.size());
              ReadShareGroupStateSummaryRequestData readSummaryRequestData = new ReadShareGroupStateSummaryRequestData()
                  .setGroupId(requestData.groupId());
              topicPartitionMap.forEach((topicId, partitionSet) -> {
                  String topicName = metadataImage.topics().topicIdToNameView().get(topicId);
                  if (topicName != null) {
                      requestTopicIdToNameMapping.put(topicId, topicName);
                      readSummaryRequestData.topics().add(new ReadShareGroupStateSummaryRequestData.ReadStateSummaryData()
                          .setTopicId(topicId)
                          .setPartitions(
                              partitionSet.stream().map(
                                  partitionIndex -> new ReadShareGroupStateSummaryRequestData.PartitionData().setPartition(partitionIndex)
                              ).toList()
                          ));
                  }
              });
              return readShareGroupStateSummary(readSummaryRequestData, requestTopicIdToNameMapping, describeShareGroupOffsetsResponseTopicList);
          });
      }
  
      private CompletableFuture<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup> readShareGroupStateSummary(
          ReadShareGroupStateSummaryRequestData readSummaryRequestData,
          Map<Uuid, String> requestTopicIdToNameMapping,
          List<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseTopic> describeShareGroupOffsetsResponseTopicList
      ) {
          // If the request for the persister is empty, just complete the operation right away.
          if (readSummaryRequestData.topics().isEmpty()) {
              return CompletableFuture.completedFuture(
                  new DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup()
                      .setGroupId(readSummaryRequestData.groupId())
                      .setTopics(describeShareGroupOffsetsResponseTopicList));
          }
  
          CompletableFuture<DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup> future = new CompletableFuture<>();
          persister.readSummary(ReadShareGroupStateSummaryParameters.from(readSummaryRequestData))
              .whenComplete((result, error) -> {
                  if (error != null) {
                      log.error("Failed to read summary of the share partition");
                      future.completeExceptionally(error);
                      return;
                  }
                  if (result == null || result.topicsData() == null) {
                      log.error("Result is null for the read state summary");
                      future.completeExceptionally(new IllegalStateException("Result is null for the read state summary"));
                      return;
                  }
  
                  // Return -1 (uninitialized offset) for the situation where the persister returned an error.
                  // This is consistent with OffsetFetch for situations in which there is no offset information to fetch.
                  // It's treated as absence of data, rather than an error.
                  result.topicsData().forEach(topicData ->
                      describeShareGroupOffsetsResponseTopicList.add(new DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseTopic()
                          .setTopicId(topicData.topicId())
                          .setTopicName(requestTopicIdToNameMapping.get(topicData.topicId()))
                          .setPartitions(topicData.partitions().stream().map(
                              partitionData -> new DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponsePartition()
                                  .setPartitionIndex(partitionData.partition())
                                  .setStartOffset(partitionData.errorCode() == Errors.NONE.code() ? partitionData.startOffset() : PartitionFactory.UNINITIALIZED_START_OFFSET)
                                  .setLeaderEpoch(partitionData.errorCode() == Errors.NONE.code() ? partitionData.leaderEpoch() : PartitionFactory.DEFAULT_LEADER_EPOCH)
                          ).toList())
                      ));
  
                  future.complete(
                      new DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup()
                          .setGroupId(readSummaryRequestData.groupId())
                          .setTopics(describeShareGroupOffsetsResponseTopicList));
              });
          return future;
      }
  
      /**
       * See {@link GroupCoordinator#deleteShareGroupOffsets(AuthorizableRequestContext, DeleteShareGroupOffsetsRequestData)}.
       */
      @Override
      public CompletableFuture<DeleteShareGroupOffsetsResponseData> deleteShareGroupOffsets(
          AuthorizableRequestContext context,
          DeleteShareGroupOffsetsRequestData requestData
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(
                  DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.COORDINATOR_NOT_AVAILABLE));
          }
  
          if (metadataImage == null) {
              return CompletableFuture.completedFuture(
                  DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.COORDINATOR_NOT_AVAILABLE));
          }
  
          String groupId = requestData.groupId();
  
          if (!isGroupIdNotEmpty(groupId)) {
              return CompletableFuture.completedFuture(
                  DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.INVALID_GROUP_ID));
          }
  
          if (requestData.topics() == null || requestData.topics().isEmpty()) {
              return CompletableFuture.completedFuture(
                  new DeleteShareGroupOffsetsResponseData()
              );
          }
  
          return runtime.scheduleWriteOperation(
              "initiate-delete-share-group-offsets",
              topicPartitionFor(groupId),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.initiateDeleteShareGroupOffsets(groupId, requestData)
          )
              .thenCompose(resultHolder -> deleteShareGroupOffsetsState(groupId, resultHolder))
              .exceptionally(exception -> handleOperationException(
                  "initiate-delete-share-group-offsets",
                  groupId,
                  exception,
                  (error, __) -> DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(error),
                  log
              ));
      }
  
      private CompletableFuture<DeleteShareGroupOffsetsResponseData> deleteShareGroupOffsetsState(
          String groupId,
          GroupCoordinatorShard.DeleteShareGroupOffsetsResultHolder resultHolder
      ) {
          if (resultHolder == null) {
              log.error("Failed to retrieve deleteState request parameters from group coordinator for the group {}", groupId);
              return CompletableFuture.completedFuture(
                  DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.UNKNOWN_SERVER_ERROR)
              );
          }
  
          if (resultHolder.topLevelErrorCode() != Errors.NONE.code()) {
              return CompletableFuture.completedFuture(
                  DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(
                      resultHolder.topLevelErrorCode(),
                      resultHolder.topLevelErrorMessage()
                  )
              );
          }
  
          List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic> errorTopicResponseList =
              resultHolder.errorTopicResponseList() == null ? new ArrayList<>() : new ArrayList<>(resultHolder.errorTopicResponseList());
  
          if (resultHolder.deleteStateRequestParameters() == null) {
              return CompletableFuture.completedFuture(
                  new DeleteShareGroupOffsetsResponseData()
                      .setResponses(errorTopicResponseList)
              );
          }
  
          return persister.deleteState(resultHolder.deleteStateRequestParameters())
              .thenCompose(result -> handleDeleteShareGroupOffsetStateResult(groupId, result, errorTopicResponseList))
              .exceptionally(throwable -> {
                  log.error("Failed to delete share group state due to: {}", throwable.getMessage(), throwable);
                  return DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.forException(throwable));
              });
      }
  
      private CompletableFuture<DeleteShareGroupOffsetsResponseData> handleDeleteShareGroupOffsetStateResult(
          String groupId,
          DeleteShareGroupStateResult result,
          List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic> errorTopicResponses
      ) {
          if (result == null || result.topicsData() == null) {
              log.error("Result is null for the delete share group state");
              Exception exception = new IllegalStateException("Result is null for the delete share group state");
              return CompletableFuture.completedFuture(
                  DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(Errors.forException(exception))
              );
          }
          Map<Uuid, String> successTopics = new HashMap<>();
          result.topicsData().forEach(topicData -> {
              Optional<PartitionErrorData> errItem = topicData.partitions().stream()
                  .filter(errData -> errData.errorCode() != Errors.NONE.code())
                  .findAny();
  
              if (errItem.isPresent()) {
                  errorTopicResponses.add(
                      new DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic()
                          .setTopicId(topicData.topicId())
                          .setTopicName(metadataImage.topics().topicIdToNameView().get(topicData.topicId()))
                          .setErrorMessage(Errors.forCode(errItem.get().errorCode()).message())
                          .setErrorCode(errItem.get().errorCode())
                  );
              } else {
                  successTopics.put(
                      topicData.topicId(),
                      metadataImage.topics().topicIdToNameView().get(topicData.topicId())
                  );
              }
          });
  
          // If there are no topics for which persister delete state request succeeded, then we can return directly from here
          if (successTopics.isEmpty()) {
              return CompletableFuture.completedFuture(
                  new DeleteShareGroupOffsetsResponseData()
                      .setResponses(errorTopicResponses)
              );
          }
  
          return completeDeleteShareGroupOffsets(groupId, successTopics, errorTopicResponses);
      }
  
      private CompletableFuture<DeleteShareGroupOffsetsResponseData> completeDeleteShareGroupOffsets(
          String groupId,
          Map<Uuid, String> successTopics,
          List<DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsResponseTopic> errorTopicResponses
      ) {
          return runtime.scheduleWriteOperation(
              "complete-delete-share-group-offsets",
              topicPartitionFor(groupId),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.completeDeleteShareGroupOffsets(groupId, successTopics, errorTopicResponses)
          ).exceptionally(exception -> handleOperationException(
              "complete-delete-share-group-offsets",
              groupId,
              exception,
              (error, __) -> DeleteShareGroupOffsetsRequest.getErrorDeleteResponseData(error),
              log
          ));
      }
  
      /**
       * See {@link GroupCoordinator#commitOffsets(AuthorizableRequestContext, OffsetCommitRequestData, BufferSupplier)}.
       */
      @Override
      public CompletableFuture<OffsetCommitResponseData> commitOffsets(
          AuthorizableRequestContext context,
          OffsetCommitRequestData request,
          BufferSupplier bufferSupplier
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(OffsetCommitRequest.getErrorResponse(
                  request,
                  Errors.COORDINATOR_NOT_AVAILABLE
              ));
          }
  
          // For backwards compatibility, we support offset commits for the empty groupId.
          if (request.groupId() == null) {
              return CompletableFuture.completedFuture(OffsetCommitRequest.getErrorResponse(
                  request,
                  Errors.INVALID_GROUP_ID
              ));
          }
  
          return runtime.scheduleWriteOperation(
              "commit-offset",
              topicPartitionFor(request.groupId()),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.commitOffset(context, request)
          ).exceptionally(exception -> handleOperationException(
              "commit-offset",
              request,
              exception,
              (error, __) -> OffsetCommitRequest.getErrorResponse(request, error),
              log
          ));
      }
  
      /**
       * See {@link GroupCoordinator#commitTransactionalOffsets(AuthorizableRequestContext, TxnOffsetCommitRequestData, BufferSupplier)}.
       */
      @Override
      public CompletableFuture<TxnOffsetCommitResponseData> commitTransactionalOffsets(
          AuthorizableRequestContext context,
          TxnOffsetCommitRequestData request,
          BufferSupplier bufferSupplier
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(TxnOffsetCommitRequest.getErrorResponse(
                  request,
                  Errors.COORDINATOR_NOT_AVAILABLE
              ));
          }
  
          if (!isGroupIdNotEmpty(request.groupId())) {
              return CompletableFuture.completedFuture(TxnOffsetCommitRequest.getErrorResponse(
                  request,
                  Errors.INVALID_GROUP_ID
              ));
          }
  
          return runtime.scheduleTransactionalWriteOperation(
              "txn-commit-offset",
              topicPartitionFor(request.groupId()),
              request.transactionalId(),
              request.producerId(),
              request.producerEpoch(),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.commitTransactionalOffset(context, request),
              context.requestVersion()
          ).exceptionally(exception -> handleOperationException(
              "txn-commit-offset",
              request,
              exception,
              (error, __) -> TxnOffsetCommitRequest.getErrorResponse(request, error),
              log
          ));
      }
  
      /**
       * See {@link GroupCoordinator#deleteOffsets(AuthorizableRequestContext, OffsetDeleteRequestData, BufferSupplier)}.
       */
      @Override
      public CompletableFuture<OffsetDeleteResponseData> deleteOffsets(
          AuthorizableRequestContext context,
          OffsetDeleteRequestData request,
          BufferSupplier bufferSupplier
      ) {
          if (!isActive.get()) {
              return CompletableFuture.completedFuture(new OffsetDeleteResponseData()
                  .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
              );
          }
  
          if (!isGroupIdNotEmpty(request.groupId())) {
              return CompletableFuture.completedFuture(new OffsetDeleteResponseData()
                  .setErrorCode(Errors.INVALID_GROUP_ID.code())
              );
          }
  
          return runtime.scheduleWriteOperation(
              "delete-offsets",
              topicPartitionFor(request.groupId()),
              Duration.ofMillis(config.offsetCommitTimeoutMs()),
              coordinator -> coordinator.deleteOffsets(context, request)
          ).exceptionally(exception -> handleOperationException(
              "delete-offsets",
              request,
              exception,
              (error, __) -> new OffsetDeleteResponseData().setErrorCode(error.code()),
              log
          ));
      }
  
      /**
       * See {@link GroupCoordinator#completeTransaction(TopicPartition, long, short, int, TransactionResult, Duration)}.
       */
      @Override
      public CompletableFuture<Void> completeTransaction(
          TopicPartition tp,
          long producerId,
          short producerEpoch,
          int coordinatorEpoch,
          TransactionResult result,
          Duration timeout
      ) {
          if (!isActive.get()) {
              return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
          }
  
          if (!tp.topic().equals(Topic.GROUP_METADATA_TOPIC_NAME)) {
              return FutureUtils.failedFuture(new IllegalStateException(
                  "Completing a transaction for " + tp + " is not expected"
              ));
          }
  
          return runtime.scheduleTransactionCompletion(
              "write-txn-marker",
              tp,
              producerId,
              producerEpoch,
              coordinatorEpoch,
              result,
              timeout
          );
      }
  
      /**
       * See {@link GroupCoordinator#onPartitionsDeleted(List, BufferSupplier)}.
       */
      @Override
      public void onPartitionsDeleted(
          List<TopicPartition> topicPartitions,
          BufferSupplier bufferSupplier
      ) throws ExecutionException, InterruptedException {
          throwIfNotActive();
  
          CompletableFuture.allOf(
              FutureUtils.mapExceptionally(
                  runtime.scheduleWriteAllOperation(
                      "on-partition-deleted",
                      Duration.ofMillis(config.offsetCommitTimeoutMs()),
                      coordinator -> coordinator.onPartitionsDeleted(topicPartitions)
                  ),
                  exception -> {
                      log.error("Could not delete offsets for deleted partitions {} due to: {}.",
                          topicPartitions, exception.getMessage(), exception
                      );
                      return null;
                  }
              ).toArray(new CompletableFuture<?>[0])
          ).get();
  
          // At this point the metadata will not have been updated
          // with the deleted topics.
          Set<Uuid> topicIds = topicPartitions.stream()
              .map(tp -> metadataImage.topics().getTopic(tp.topic()).id())
              .collect(Collectors.toSet());
  
          CompletableFuture.allOf(
              FutureUtils.mapExceptionally(
                  runtime.scheduleWriteAllOperation(
                      "maybe-cleanup-share-group-state",
                      Duration.ofMillis(config.offsetCommitTimeoutMs()),
                      coordinator -> coordinator.maybeCleanupShareGroupState(topicIds)
                  ),
                  exception -> {
                      log.error("Unable to cleanup state for the deleted topics {}", topicIds, exception);
                      return null;
                  }
              ).toArray(new CompletableFuture<?>[0])
          ).get();
      }
  
      /**
       * See {@link GroupCoordinator#onElection(int, int)}.
       */
      @Override
      public void onElection(
          int groupMetadataPartitionIndex,
          int groupMetadataPartitionLeaderEpoch
      ) {
          throwIfNotActive();
          runtime.scheduleLoadOperation(
              new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, groupMetadataPartitionIndex),
              groupMetadataPartitionLeaderEpoch
          );
      }
  
      /**
       * See {@link GroupCoordinator#onResignation(int, OptionalInt)}.
       */
      @Override
      public void onResignation(
          int groupMetadataPartitionIndex,
          OptionalInt groupMetadataPartitionLeaderEpoch
      ) {
          throwIfNotActive();
          runtime.scheduleUnloadOperation(
              new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, groupMetadataPartitionIndex),
              groupMetadataPartitionLeaderEpoch
          );
      }
  
      /**
       * See {@link GroupCoordinator#onNewMetadataImage(MetadataImage, MetadataDelta)}.
       */
      @Override
      public void onNewMetadataImage(
          MetadataImage newImage,
          MetadataDelta delta
      ) {
          throwIfNotActive();
          metadataImage = newImage;
          runtime.onNewMetadataImage(newImage, delta);
      }
  
      /**
       * See {@link GroupCoordinator#groupMetadataTopicConfigs()}.
       */
      @Override
      public Properties groupMetadataTopicConfigs() {
          Properties properties = new Properties();
          properties.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
          properties.put(TopicConfig.COMPRESSION_TYPE_CONFIG, BrokerCompressionType.PRODUCER.name);
          properties.put(TopicConfig.SEGMENT_BYTES_CONFIG, String.valueOf(config.offsetsTopicSegmentBytes()));
          return properties;
      }
  
      /**
       * See {@link GroupCoordinator#groupConfig(String)}.
       */
      @Override
      public Optional<GroupConfig> groupConfig(String groupId) {
          return groupConfigManager.groupConfig(groupId);
      }
  
      /**
       * See {@link GroupCoordinator#updateGroupConfig(String, Properties)}.
       */
      @Override
      public void updateGroupConfig(String groupId, Properties newGroupConfig) {
          groupConfigManager.updateGroupConfig(groupId, newGroupConfig);
      }
  
      /**
       * See {@link GroupCoordinator#startup(IntSupplier)}.
       */
      @Override
      public void startup(
          IntSupplier groupMetadataTopicPartitionCount
      ) {
          if (!isActive.compareAndSet(false, true)) {
              log.warn("Group coordinator is already running.");
              return;
          }
  
          log.info("Starting up.");
          numPartitions = groupMetadataTopicPartitionCount.getAsInt();
          isActive.set(true);
          log.info("Startup complete.");
      }
  
      /**
       * See {@link GroupCoordinator#shutdown()}.
       */
      @Override
      public void shutdown() {
          if (!isActive.compareAndSet(true, false)) {
              log.warn("Group coordinator is already shutting down.");
              return;
          }
  
          log.info("Shutting down.");
          isActive.set(false);
          Utils.closeQuietly(runtime, "coordinator runtime");
          Utils.closeQuietly(groupCoordinatorMetrics, "group coordinator metrics");
          Utils.closeQuietly(groupConfigManager, "group config manager");
          log.info("Shutdown complete.");
      }
  
      private static boolean isGroupIdNotEmpty(String groupId) {
          return groupId != null && !groupId.isEmpty();
      }
  
      /**
       * This is the handler used by offset fetch operations to convert errors to coordinator errors.
       * The handler also handles and logs unexpected errors.
       *
       * @param operationName     The name of the operation.
       * @param request           The OffsetFetchRequestGroup request.
       * @param exception         The exception to handle.
       * @return The OffsetFetchRequestGroup response.
       */
      private OffsetFetchResponseData.OffsetFetchResponseGroup handleOffsetFetchException(
          String operationName,
          OffsetFetchRequestData.OffsetFetchRequestGroup request,
          Throwable exception
      ) {
          ApiError apiError = ApiError.fromThrowable(exception);
  
          switch (apiError.error()) {
              case UNKNOWN_TOPIC_OR_PARTITION:
              case NOT_ENOUGH_REPLICAS:
              case REQUEST_TIMED_OUT:
                  // Remap REQUEST_TIMED_OUT to NOT_COORDINATOR, since consumers on versions prior
                  // to 3.9 do not expect the error and won't retry the request. NOT_COORDINATOR
                  // additionally triggers coordinator re-lookup, which is necessary if the client is
                  // talking to a zombie coordinator.
                  //
                  // While handleOperationException does remap UNKNOWN_TOPIC_OR_PARTITION,
                  // NOT_ENOUGH_REPLICAS and REQUEST_TIMED_OUT to COORDINATOR_NOT_AVAILABLE,
                  // COORDINATOR_NOT_AVAILABLE is also not handled by consumers on versions prior to
                  // 3.9.
                  return new OffsetFetchResponseData.OffsetFetchResponseGroup()
                      .setGroupId(request.groupId())
                      .setErrorCode(Errors.NOT_COORDINATOR.code());
  
              default:
                  return handleOperationException(
                      operationName,
                      request,
                      exception,
                      (error, __) -> new OffsetFetchResponseData.OffsetFetchResponseGroup()
                          .setGroupId(request.groupId())
                          .setErrorCode(error.code()),
                      log
                  );
          }
      }
  
      private static void requireNonNull(Object obj, String msg) {
          if (obj == null) {
              throw new IllegalArgumentException(msg);
          }
      }
  }
  