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.clients.consumer.internals;
  
  import org.apache.kafka.clients.CommonClientConfigs;
  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
  import org.apache.kafka.clients.consumer.internals.events.ErrorEvent;
  import org.apache.kafka.clients.consumer.internals.metrics.HeartbeatMetricsManager;
  import org.apache.kafka.common.TopicPartition;
  import org.apache.kafka.common.errors.GroupAuthorizationException;
  import org.apache.kafka.common.errors.RetriableException;
  import org.apache.kafka.common.errors.UnsupportedVersionException;
  import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
  import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
  import org.apache.kafka.common.metrics.Metrics;
  import org.apache.kafka.common.protocol.Errors;
  import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
  import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse;
  import org.apache.kafka.common.utils.LogContext;
  import org.apache.kafka.common.utils.Time;
  import org.apache.kafka.common.utils.Timer;
  
  import org.slf4j.Logger;
  
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.Comparator;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Set;
  import java.util.stream.Collectors;
  import java.util.stream.IntStream;
  
  import static org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult.EMPTY;
  
  /**
   * <p>Manages the request creation and response handling for the streams group heartbeat. The class creates a
   * heartbeat request using the state stored in the membership manager. The requests can be retrieved
   * by calling {@link StreamsGroupHeartbeatRequestManager#poll(long)}. Once the response is received, it updates the
   * state in the membership manager and handles any errors.
   *
   * <p>The heartbeat manager generates heartbeat requests based on the member state. It's also responsible
   * for the timing of the heartbeat requests to ensure they are sent according to the heartbeat interval
   * (while the member state is stable) or on demand (while the member is acknowledging an assignment or
   * leaving the group).
   */
  public class StreamsGroupHeartbeatRequestManager implements RequestManager {
  
      private static final String UNSUPPORTED_VERSION_ERROR_MESSAGE = "The cluster does not support the STREAMS group " +
          "protocol or does not support the versions of the STREAMS group protocol used by this client " +
          "(used versions: " + StreamsGroupHeartbeatRequestData.LOWEST_SUPPORTED_VERSION + " to " +
          StreamsGroupHeartbeatRequestData.HIGHEST_SUPPORTED_VERSION + ").";
  
      static class HeartbeatState {
  
          // Fields of StreamsGroupHeartbeatRequest sent in the most recent request
          static class LastSentFields {
  
              private StreamsRebalanceData.Assignment assignment = null;
  
              LastSentFields() {
              }
  
              void reset() {
                  assignment = null;
              }
          }
  
          private final StreamsMembershipManager membershipManager;
          private final int rebalanceTimeoutMs;
          private final StreamsRebalanceData streamsRebalanceData;
          private final LastSentFields lastSentFields = new LastSentFields();
          private int endpointInformationEpoch = -1;
  
  
          public HeartbeatState(final StreamsRebalanceData streamsRebalanceData,
                                final StreamsMembershipManager membershipManager,
                                final int rebalanceTimeoutMs) {
              this.membershipManager = membershipManager;
              this.streamsRebalanceData = streamsRebalanceData;
              this.rebalanceTimeoutMs = rebalanceTimeoutMs;
          }
  
          public void reset() {
              lastSentFields.reset();
          }
  
          public int endpointInformationEpoch() {
              return endpointInformationEpoch;
          }
  
          public void setEndpointInformationEpoch(int endpointInformationEpoch) {
              this.endpointInformationEpoch = endpointInformationEpoch;
          }
  
          public StreamsGroupHeartbeatRequestData buildRequestData() {
              StreamsGroupHeartbeatRequestData data = new StreamsGroupHeartbeatRequestData();
              data.setGroupId(membershipManager.groupId());
              data.setMemberId(membershipManager.memberId());
              data.setMemberEpoch(membershipManager.memberEpoch());
              data.setEndpointInformationEpoch(endpointInformationEpoch);
              membershipManager.groupInstanceId().ifPresent(data::setInstanceId);
  
              boolean joining = membershipManager.state() == MemberState.JOINING;
  
              if (joining) {
                  StreamsGroupHeartbeatRequestData.Topology topology = new StreamsGroupHeartbeatRequestData.Topology();
                  topology.setSubtopologies(fromStreamsToHeartbeatRequest(streamsRebalanceData.subtopologies()));
                  topology.setEpoch(streamsRebalanceData.topologyEpoch());
                  data.setTopology(topology);
                  data.setRebalanceTimeoutMs(rebalanceTimeoutMs);
                  data.setProcessId(streamsRebalanceData.processId().toString());
                  streamsRebalanceData.endpoint().ifPresent(userEndpoint -> {
                      data.setUserEndpoint(new StreamsGroupHeartbeatRequestData.Endpoint()
                          .setHost(userEndpoint.host())
                          .setPort(userEndpoint.port())
                      );
                  });
                  data.setClientTags(streamsRebalanceData.clientTags().entrySet().stream()
                      .map(entry -> new StreamsGroupHeartbeatRequestData.KeyValue()
                          .setKey(entry.getKey())
                          .setValue(entry.getValue())
                      )
                      .collect(Collectors.toList()));
                  data.setActiveTasks(fromStreamsToHeartbeatRequest(Set.of()));
                  data.setStandbyTasks(fromStreamsToHeartbeatRequest(Set.of()));
                  data.setWarmupTasks(fromStreamsToHeartbeatRequest(Set.of()));
              } else {
                  StreamsRebalanceData.Assignment reconciledAssignment = streamsRebalanceData.reconciledAssignment();
                  if (!reconciledAssignment.equals(lastSentFields.assignment)) {
                      data.setActiveTasks(fromStreamsToHeartbeatRequest(reconciledAssignment.activeTasks()));
                      data.setStandbyTasks(fromStreamsToHeartbeatRequest(reconciledAssignment.standbyTasks()));
                      data.setWarmupTasks(fromStreamsToHeartbeatRequest(reconciledAssignment.warmupTasks()));
                      lastSentFields.assignment = reconciledAssignment;
                  }
              }
              data.setShutdownApplication(streamsRebalanceData.shutdownRequested());
              return data;
          }
  
          private static List<StreamsGroupHeartbeatRequestData.TaskIds> fromStreamsToHeartbeatRequest(final Set<StreamsRebalanceData.TaskId> tasks) {
              return tasks.stream()
                  .collect(
                      Collectors.groupingBy(StreamsRebalanceData.TaskId::subtopologyId,
                          Collectors.mapping(StreamsRebalanceData.TaskId::partitionId, Collectors.toList()))
                  )
                  .entrySet()
                  .stream()
                  .map(entry -> {
                      return new StreamsGroupHeartbeatRequestData.TaskIds()
                          .setSubtopologyId(entry.getKey())
                          .setPartitions(entry.getValue());
                  })
                  .collect(Collectors.toList());
          }
  
          private static List<StreamsGroupHeartbeatRequestData.Subtopology> fromStreamsToHeartbeatRequest(final Map<String, StreamsRebalanceData.Subtopology> subtopologies) {
              final List<StreamsGroupHeartbeatRequestData.Subtopology> subtopologiesForRequest = new ArrayList<>(subtopologies.size());
              for (final Map.Entry<String, StreamsRebalanceData.Subtopology> subtopology : subtopologies.entrySet()) {
                  subtopologiesForRequest.add(fromStreamsToHeartbeatRequest(subtopology.getKey(), subtopology.getValue()));
              }
              subtopologiesForRequest.sort(Comparator.comparing(StreamsGroupHeartbeatRequestData.Subtopology::subtopologyId));
              return subtopologiesForRequest;
          }
  
          private static StreamsGroupHeartbeatRequestData.Subtopology fromStreamsToHeartbeatRequest(final String subtopologyId,
                                                                                                    final StreamsRebalanceData.Subtopology subtopology) {
              final StreamsGroupHeartbeatRequestData.Subtopology subtopologyData = new StreamsGroupHeartbeatRequestData.Subtopology();
              subtopologyData.setSubtopologyId(subtopologyId);
              ArrayList<String> sortedSourceTopics = new ArrayList<>(subtopology.sourceTopics());
              Collections.sort(sortedSourceTopics);
              subtopologyData.setSourceTopics(sortedSourceTopics);
              ArrayList<String> sortedSinkTopics = new ArrayList<>(subtopology.repartitionSinkTopics());
              Collections.sort(sortedSinkTopics);
              subtopologyData.setRepartitionSinkTopics(sortedSinkTopics);
              subtopologyData.setRepartitionSourceTopics(getRepartitionTopicsInfoFromStreams(subtopology));
              subtopologyData.setStateChangelogTopics(getChangelogTopicsInfoFromStreams(subtopology));
              subtopologyData.setCopartitionGroups(
                  getCopartitionGroupsFromStreams(subtopology.copartitionGroups(), subtopologyData));
              return subtopologyData;
          }
  
          private static List<StreamsGroupHeartbeatRequestData.CopartitionGroup> getCopartitionGroupsFromStreams(final Collection<Set<String>> copartitionGroups,
                                                                                                                 final StreamsGroupHeartbeatRequestData.Subtopology subtopologyData) {
              final Map<String, Short> sourceTopicsMap =
                  IntStream.range(0, subtopologyData.sourceTopics().size())
                      .boxed()
                      .collect(Collectors.toMap(subtopologyData.sourceTopics()::get, Integer::shortValue));
  
              final Map<String, Short> repartitionSourceTopics =
                  IntStream.range(0, subtopologyData.repartitionSourceTopics().size())
                      .boxed()
                      .collect(
                          Collectors.toMap(x -> subtopologyData.repartitionSourceTopics().get(x).name(),
                              Integer::shortValue));
  
              return copartitionGroups.stream()
                  .map(x -> getCopartitionGroupFromStreams(x, sourceTopicsMap, repartitionSourceTopics))
                  .collect(Collectors.toList());
          }
  
          private static StreamsGroupHeartbeatRequestData.CopartitionGroup getCopartitionGroupFromStreams(final Set<String> topicNames,
                                                                                                          final Map<String, Short> sourceTopicsMap,
                                                                                                          final Map<String, Short> repartitionSourceTopics) {
              StreamsGroupHeartbeatRequestData.CopartitionGroup copartitionGroup = new StreamsGroupHeartbeatRequestData.CopartitionGroup();
  
              topicNames.forEach(topicName -> {
                  if (sourceTopicsMap.containsKey(topicName)) {
                      copartitionGroup.sourceTopics().add(sourceTopicsMap.get(topicName));
                  } else if (repartitionSourceTopics.containsKey(topicName)) {
                      copartitionGroup.repartitionSourceTopics()
                          .add(repartitionSourceTopics.get(topicName));
                  } else {
                      throw new IllegalStateException(
                          "Source topic not found in subtopology: " + topicName);
                  }
              });
  
              return copartitionGroup;
          }
  
          private static List<StreamsGroupHeartbeatRequestData.TopicInfo> getRepartitionTopicsInfoFromStreams(final StreamsRebalanceData.Subtopology subtopologyDataFromStreams) {
              final List<StreamsGroupHeartbeatRequestData.TopicInfo> repartitionTopicsInfo = new ArrayList<>();
              for (final Map.Entry<String, StreamsRebalanceData.TopicInfo> repartitionTopic : subtopologyDataFromStreams.repartitionSourceTopics().entrySet()) {
                  final StreamsGroupHeartbeatRequestData.TopicInfo repartitionTopicInfo = new StreamsGroupHeartbeatRequestData.TopicInfo();
                  repartitionTopicInfo.setName(repartitionTopic.getKey());
                  repartitionTopic.getValue().numPartitions().ifPresent(repartitionTopicInfo::setPartitions);
                  repartitionTopic.getValue().replicationFactor().ifPresent(repartitionTopicInfo::setReplicationFactor);
                  repartitionTopic.getValue().topicConfigs().forEach((k, v) ->
                      repartitionTopicInfo.topicConfigs().add(new StreamsGroupHeartbeatRequestData.KeyValue().setKey(k).setValue(v))
                  );
                  repartitionTopicsInfo.add(repartitionTopicInfo);
                  repartitionTopicInfo.topicConfigs().sort(Comparator.comparing(StreamsGroupHeartbeatRequestData.KeyValue::key));
              }
              repartitionTopicsInfo.sort(Comparator.comparing(StreamsGroupHeartbeatRequestData.TopicInfo::name));
              return repartitionTopicsInfo;
          }
  
          private static List<StreamsGroupHeartbeatRequestData.TopicInfo> getChangelogTopicsInfoFromStreams(final StreamsRebalanceData.Subtopology subtopologyDataFromStreams) {
              final List<StreamsGroupHeartbeatRequestData.TopicInfo> changelogTopicsInfo = new ArrayList<>();
              for (final Map.Entry<String, StreamsRebalanceData.TopicInfo> changelogTopic : subtopologyDataFromStreams.stateChangelogTopics().entrySet()) {
                  final StreamsGroupHeartbeatRequestData.TopicInfo changelogTopicInfo = new StreamsGroupHeartbeatRequestData.TopicInfo();
                  changelogTopicInfo.setName(changelogTopic.getKey());
                  changelogTopic.getValue().replicationFactor().ifPresent(changelogTopicInfo::setReplicationFactor);
                  changelogTopic.getValue().topicConfigs().forEach((k, v) ->
                      changelogTopicInfo.topicConfigs().add(new StreamsGroupHeartbeatRequestData.KeyValue().setKey(k).setValue(v))
                  );
                  changelogTopicInfo.topicConfigs().sort(Comparator.comparing(StreamsGroupHeartbeatRequestData.KeyValue::key));
                  changelogTopicsInfo.add(changelogTopicInfo);
              }
              changelogTopicsInfo.sort(Comparator.comparing(StreamsGroupHeartbeatRequestData.TopicInfo::name));
              return changelogTopicsInfo;
          }
      }
  
      private final Logger logger;
  
      private final int maxPollIntervalMs;
  
      private final CoordinatorRequestManager coordinatorRequestManager;
  
      private final HeartbeatRequestState heartbeatRequestState;
  
      private final HeartbeatState heartbeatState;
  
      private final StreamsMembershipManager membershipManager;
  
      private final BackgroundEventHandler backgroundEventHandler;
  
      private final HeartbeatMetricsManager metricsManager;
  
      private StreamsRebalanceData streamsRebalanceData;
  
      /**
       * Timer for tracking the time since the last consumer poll.  If the timer expires, the consumer will stop
       * sending heartbeat until the next poll.
       */
      private final Timer pollTimer;
  
      public StreamsGroupHeartbeatRequestManager(final LogContext logContext,
                                                 final Time time,
                                                 final ConsumerConfig config,
                                                 final CoordinatorRequestManager coordinatorRequestManager,
                                                 final StreamsMembershipManager membershipManager,
                                                 final BackgroundEventHandler backgroundEventHandler,
                                                 final Metrics metrics,
                                                 final StreamsRebalanceData streamsRebalanceData) {
          this.logger = logContext.logger(getClass());
          this.coordinatorRequestManager = Objects.requireNonNull(
              coordinatorRequestManager,
              "Coordinator request manager cannot be null"
          );
          this.membershipManager = Objects.requireNonNull(
              membershipManager,
              "Streams membership manager cannot be null"
          );
          this.backgroundEventHandler = Objects.requireNonNull(
              backgroundEventHandler,
              "Background event handler cannot be null"
          );
          this.metricsManager = new HeartbeatMetricsManager(
              Objects.requireNonNull(metrics, "Metrics cannot be null")
          );
          this.streamsRebalanceData = Objects.requireNonNull(streamsRebalanceData, "Streams rebalance data cannot be null");
          this.maxPollIntervalMs = config.getInt(CommonClientConfigs.MAX_POLL_INTERVAL_MS_CONFIG);
          long retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
          long retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
          this.heartbeatState = new HeartbeatState(streamsRebalanceData, membershipManager, maxPollIntervalMs);
          this.heartbeatRequestState = new HeartbeatRequestState(
              logContext,
              time,
              0,
              retryBackoffMs,
              retryBackoffMaxMs,
              maxPollIntervalMs
          );
          this.pollTimer = time.timer(maxPollIntervalMs);
      }
  
      /**
       * This will build a heartbeat request if one must be sent, determined based on the member
       * state. A heartbeat is sent when all of the following applies:
       * <ol>
       *     <li>Member is part of the consumer group or wants to join it.</li>
       *     <li>The heartbeat interval has expired, or the member is in a state that indicates
       *     that it should heartbeat without waiting for the interval.</li>
       * </ol>
       * This will also determine the maximum wait time until the next poll based on the member's
       * state.
       * <ol>
       *     <li>If the member is without a coordinator or is in a failed state, the timer is set
       *     to Long.MAX_VALUE, as there's no need to send a heartbeat.</li>
       *     <li>If the member cannot send a heartbeat due to either exponential backoff, it will
       *     return the remaining time left on the backoff timer.</li>
       *     <li>If the member's heartbeat timer has not expired, It will return the remaining time
       *     left on the heartbeat timer.</li>
       *     <li>If the member can send a heartbeat, the timer is set to the current heartbeat interval.</li>
       * </ol>
       *
       * @return {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} that includes a
       *         heartbeat request if one must be sent, and the time to wait until the next poll.
       */
      @Override
      public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
          if (coordinatorRequestManager.coordinator().isEmpty() || membershipManager.shouldSkipHeartbeat()) {
              membershipManager.onHeartbeatRequestSkipped();
              maybePropagateCoordinatorFatalErrorEvent();
              return NetworkClientDelegate.PollResult.EMPTY;
          }
          pollTimer.update(currentTimeMs);
          if (pollTimer.isExpired() && !membershipManager.isLeavingGroup()) {
              logger.warn("Consumer poll timeout has expired. This means the time between " +
                  "subsequent calls to poll() was longer than the configured max.poll.interval.ms, " +
                  "which typically implies that the poll loop is spending too much time processing " +
                  "messages. You can address this either by increasing max.poll.interval.ms or by " +
                  "reducing the maximum size of batches returned in poll() with max.poll.records.");
  
              membershipManager.onPollTimerExpired();
              NetworkClientDelegate.UnsentRequest leaveHeartbeat = makeHeartbeatRequestAndLogResponse(currentTimeMs);
  
              // We can ignore the leave response because we can join before or after receiving the response.
              heartbeatRequestState.reset();
              heartbeatState.reset();
              return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs(), Collections.singletonList(leaveHeartbeat));
          }
          if (shouldHeartbeatBeforeIntervalExpires() || heartbeatRequestState.canSendRequest(currentTimeMs)) {
              NetworkClientDelegate.UnsentRequest request = makeHeartbeatRequestAndHandleResponse(currentTimeMs);
              return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs(), Collections.singletonList(request));
          } else {
              return new NetworkClientDelegate.PollResult(heartbeatRequestState.timeToNextHeartbeatMs(currentTimeMs));
          }
      }
  
      /**
       * Generate a heartbeat request to leave the group if the state is still LEAVING when this is
       * called to close the consumer.
       * <p/>
       * Note that when closing the consumer, even though an event to Unsubscribe is generated
       * (triggers callbacks and sends leave group), it could be the case that the Unsubscribe event
       * processing does not complete in time and moves on to close the managers (ex. calls to
       * close with zero timeout). So we could end up on this pollOnClose with the member in
       * {@link MemberState#PREPARE_LEAVING} (ex. app thread did not have the time to process the
       * event to execute callbacks), or {@link MemberState#LEAVING} (ex. the leave request could
       * not be sent due to coordinator not available at that time). In all cases, the pollOnClose
       * will be triggered right before sending the final requests, so we ensure that we generate
       * the request to leave if needed.
       *
       * @param currentTimeMs The current system time in milliseconds at which the method was called
       * @return PollResult containing the request to send
       */
      @Override
      public NetworkClientDelegate.PollResult pollOnClose(long currentTimeMs) {
          if (membershipManager.isLeavingGroup()) {
              NetworkClientDelegate.UnsentRequest request = makeHeartbeatRequestAndLogResponse(currentTimeMs);
              return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs(), List.of(request));
          }
          return EMPTY;
      }
  
      public StreamsMembershipManager membershipManager() {
          return membershipManager;
      }
  
      /**
       * Returns the delay for which the application thread can safely wait before it should be responsive
       * to results from the request managers. For example, the subscription state can change when heartbeats
       * are sent, so blocking for longer than the heartbeat interval might mean the application thread is not
       * responsive to changes.
       *
       * <p>Similarly, we may have to unblock the application thread to send a `PollApplicationEvent` to make sure
       * our poll timer will not expire while we are polling.
       *
       * <p>In the event that heartbeats are currently being skipped, this still returns the next heartbeat
       * delay rather than {@code Long.MAX_VALUE} so that the application thread remains responsive.
       */
      @Override
      public long maximumTimeToWait(long currentTimeMs) {
          pollTimer.update(currentTimeMs);
          if (pollTimer.isExpired() ||
              membershipManager.shouldNotWaitForHeartbeatInterval() && !heartbeatRequestState.requestInFlight()) {
  
              return 0L;
          }
          return Math.min(pollTimer.remainingMs() / 2, heartbeatRequestState.timeToNextHeartbeatMs(currentTimeMs));
      }
  
      public void resetPollTimer(final long pollMs) {
          pollTimer.update(pollMs);
          if (pollTimer.isExpired()) {
              logger.warn("Time between subsequent calls to poll() was longer than the configured " +
                      "max.poll.interval.ms, exceeded approximately by {} ms. Member {} will rejoin the group now.",
                  pollTimer.isExpiredBy(), membershipManager.memberId());
              membershipManager.maybeRejoinStaleMember();
          }
          pollTimer.reset(maxPollIntervalMs);
      }
  
      /**
       * A heartbeat should be sent without waiting for the heartbeat interval to expire if:
       * - the member is leaving the group
       * or
       * - the member is joining the group or acknowledging the assignment and for both cases there is no heartbeat request
       *   in flight.
       *
       * @return true if a heartbeat should be sent before the interval expires, false otherwise
       */
      private boolean shouldHeartbeatBeforeIntervalExpires() {
          return membershipManager.state() == MemberState.LEAVING
              ||
              (membershipManager.state() == MemberState.JOINING || membershipManager.state() == MemberState.ACKNOWLEDGING)
                  && !heartbeatRequestState.requestInFlight();
      }
  
      private void maybePropagateCoordinatorFatalErrorEvent() {
          coordinatorRequestManager.getAndClearFatalError()
              .ifPresent(fatalError -> backgroundEventHandler.add(new ErrorEvent(fatalError)));
      }
  
      private NetworkClientDelegate.UnsentRequest makeHeartbeatRequestAndLogResponse(final long currentTimeMs) {
          return makeHeartbeatRequest(currentTimeMs).whenComplete((response, exception) -> {
              if (response != null) {
                  metricsManager.recordRequestLatency(response.requestLatencyMs());
                  Errors error = Errors.forCode(((StreamsGroupHeartbeatResponse) response.responseBody()).data().errorCode());
                  if (error == Errors.NONE)
                      logger.debug("StreamsGroupHeartbeatRequest responded successfully: {}", response);
                  else
                      logger.error("StreamsGroupHeartbeatRequest failed because of {}: {}", error, response);
              } else {
                  logger.error("StreamsGroupHeartbeatRequest failed because of unexpected exception.", exception);
              }
          });
      }
  
      private NetworkClientDelegate.UnsentRequest makeHeartbeatRequestAndHandleResponse(final long currentTimeMs) {
          NetworkClientDelegate.UnsentRequest request = makeHeartbeatRequest(currentTimeMs);
          return request.whenComplete((response, exception) -> {
              long completionTimeMs = request.handler().completionTimeMs();
              if (response != null) {
                  metricsManager.recordRequestLatency(response.requestLatencyMs());
                  onResponse((StreamsGroupHeartbeatResponse) response.responseBody(), completionTimeMs);
              } else {
                  onFailure(exception, completionTimeMs);
              }
          });
      }
  
      private NetworkClientDelegate.UnsentRequest makeHeartbeatRequest(final long currentTimeMs) {
          NetworkClientDelegate.UnsentRequest request = new NetworkClientDelegate.UnsentRequest(
              new StreamsGroupHeartbeatRequest.Builder(this.heartbeatState.buildRequestData(), true),
              coordinatorRequestManager.coordinator()
          );
          heartbeatRequestState.onSendAttempt(currentTimeMs);
          membershipManager.onHeartbeatRequestGenerated();
          metricsManager.recordHeartbeatSentMs(currentTimeMs);
          heartbeatRequestState.resetTimer();
          return request;
      }
  
      private void onResponse(final StreamsGroupHeartbeatResponse response, long currentTimeMs) {
          if (Errors.forCode(response.data().errorCode()) == Errors.NONE) {
              onSuccessResponse(response, currentTimeMs);
          } else {
              onErrorResponse(response, currentTimeMs);
          }
      }
  
      private void onSuccessResponse(final StreamsGroupHeartbeatResponse response, final long currentTimeMs) {
          final StreamsGroupHeartbeatResponseData data = response.data();
          heartbeatRequestState.updateHeartbeatIntervalMs(data.heartbeatIntervalMs());
          heartbeatRequestState.onSuccessfulAttempt(currentTimeMs);
          heartbeatState.setEndpointInformationEpoch(data.endpointInformationEpoch());
  
          if (data.partitionsByUserEndpoint() != null) {
              streamsRebalanceData.setPartitionsByHost(convertHostInfoMap(data));
          }
  
          List<StreamsGroupHeartbeatResponseData.Status> statuses = data.status();
          if (statuses != null) {
              streamsRebalanceData.setStatuses(statuses);
              if (!statuses.isEmpty()) {
                  String statusDetails = statuses.stream()
                      .map(status -> "(" + status.statusCode() + ") " + status.statusDetail())
                      .collect(Collectors.joining(", "));
                  logger.warn("Membership is in the following statuses: {}", statusDetails);
              }
          }
  
          membershipManager.onHeartbeatSuccess(response);
      }
  
      private void onErrorResponse(final StreamsGroupHeartbeatResponse response, final long currentTimeMs) {
          final Errors error = Errors.forCode(response.data().errorCode());
          final String errorMessage = response.data().errorMessage();
  
          heartbeatState.reset();
          this.heartbeatRequestState.onFailedAttempt(currentTimeMs);
  
          switch (error) {
              case NOT_COORDINATOR:
                  logInfo(
                      String.format("StreamsGroupHeartbeatRequest failed because the group coordinator %s is incorrect. " +
                          "Will attempt to find the coordinator again and retry", coordinatorRequestManager.coordinator()),
                      response,
                      currentTimeMs
                  );
                  coordinatorRequestManager.markCoordinatorUnknown(errorMessage, currentTimeMs);
                  // Skip backoff so that the next HB is sent as soon as the new coordinator is discovered
                  heartbeatRequestState.reset();
                  break;
  
              case COORDINATOR_NOT_AVAILABLE:
                  logInfo(
                      String.format("StreamsGroupHeartbeatRequest failed because the group coordinator %s is not available. " +
                          "Will attempt to find the coordinator again and retry", coordinatorRequestManager.coordinator()),
                      response,
                      currentTimeMs
                  );
                  coordinatorRequestManager.markCoordinatorUnknown(errorMessage, currentTimeMs);
                  // Skip backoff so that the next HB is sent as soon as the new coordinator is discovered
                  heartbeatRequestState.reset();
                  break;
  
              case COORDINATOR_LOAD_IN_PROGRESS:
                  logInfo(
                      String.format("StreamsGroupHeartbeatRequest failed because the group coordinator %s is still loading. " +
                      "Will retry", coordinatorRequestManager.coordinator()),
                      response,
                      currentTimeMs
                  );
                  break;
  
              case GROUP_AUTHORIZATION_FAILED:
                  GroupAuthorizationException exception =
                      GroupAuthorizationException.forGroupId(membershipManager.groupId());
                  logger.error("StreamsGroupHeartbeatRequest failed due to group authorization failure: {}",
                      exception.getMessage());
                  handleFatalFailure(error.exception(exception.getMessage()));
                  break;
  
              case TOPIC_AUTHORIZATION_FAILED:
                  logger.error("StreamsGroupHeartbeatRequest failed for member {} with state {} due to {}: {}",
                      membershipManager.memberId(), membershipManager.state(), error, errorMessage);
                  // Propagate auth error received in HB so that it's returned on poll.
                  // Member should stay in its current state so it can recover if ever the missing ACLs are added.
                  backgroundEventHandler.add(new ErrorEvent(error.exception()));
                  break;
  
              case INVALID_REQUEST:
              case GROUP_MAX_SIZE_REACHED:
              case STREAMS_INVALID_TOPOLOGY:
              case STREAMS_INVALID_TOPOLOGY_EPOCH:
              case STREAMS_TOPOLOGY_FENCED:
                  logger.error("StreamsGroupHeartbeatRequest failed due to {}: {}", error, errorMessage);
                  handleFatalFailure(error.exception(errorMessage));
                  break;
  
              case FENCED_MEMBER_EPOCH:
                  logInfo(
                      String.format("StreamsGroupHeartbeatRequest failed for member %s because epoch %s is fenced.",
                          membershipManager.memberId(), membershipManager.memberEpoch()),
                      response,
                      currentTimeMs
                  );
                  membershipManager.onFenced();
                  // Skip backoff so that a next HB to rejoin is sent as soon as the fenced member releases its assignment
                  heartbeatRequestState.reset();
                  break;
  
              case UNKNOWN_MEMBER_ID:
                  logInfo(
                      String.format("StreamsGroupHeartbeatRequest failed because member %s is unknown.",
                          membershipManager.memberId()),
                      response,
                      currentTimeMs
                  );
                  membershipManager.onFenced();
                  // Skip backoff so that a next HB to rejoin is sent as soon as the fenced member releases its assignment
                  heartbeatRequestState.reset();
                  break;
  
              case UNSUPPORTED_VERSION:
                  logger.error("StreamsGroupHeartbeatRequest failed due to {}: {}", error, UNSUPPORTED_VERSION_ERROR_MESSAGE);
                  handleFatalFailure(error.exception(UNSUPPORTED_VERSION_ERROR_MESSAGE));
                  break;
  
              default:
                  logger.error("StreamsGroupHeartbeatRequest failed due to unexpected error {}: {}", error, errorMessage);
                  handleFatalFailure(error.exception(errorMessage));
          }
          membershipManager.onFatalHeartbeatFailure();
      }
  
      private void logInfo(final String message,
                           final StreamsGroupHeartbeatResponse response,
                           final long currentTimeMs) {
          logger.info("{} in {}ms: {}",
              message,
              heartbeatRequestState.remainingBackoffMs(currentTimeMs),
              response.data().errorMessage());
      }
  
      private void onFailure(final Throwable exception, final long responseTimeMs) {
          heartbeatRequestState.onFailedAttempt(responseTimeMs);
          heartbeatState.reset();
          if (exception instanceof RetriableException) {
              coordinatorRequestManager.handleCoordinatorDisconnect(exception, responseTimeMs);
              String message = String.format("StreamsGroupHeartbeatRequest failed because of a retriable exception. Will retry in %s ms: %s",
                  heartbeatRequestState.remainingBackoffMs(responseTimeMs),
                  exception.getMessage());
              logger.debug(message);
              membershipManager.onRetriableHeartbeatFailure();
          } else {
              if (exception instanceof UnsupportedVersionException) {
                  logger.error("StreamsGroupHeartbeatRequest failed because of an unsupported version exception: {}",
                      exception.getMessage());
                  handleFatalFailure(new UnsupportedVersionException(UNSUPPORTED_VERSION_ERROR_MESSAGE));
              } else {
                  logger.error("StreamsGroupHeartbeatRequest failed because of a fatal exception while sending request: {}",
                      exception.getMessage());
                  handleFatalFailure(exception);
              }
              membershipManager.onFatalHeartbeatFailure();
          }
      }
  
      private void handleFatalFailure(Throwable error) {
          backgroundEventHandler.add(new ErrorEvent(error));
          membershipManager.transitionToFatal();
      }
  
      private static Map<StreamsRebalanceData.HostInfo, StreamsRebalanceData.EndpointPartitions> convertHostInfoMap(
              final StreamsGroupHeartbeatResponseData data) {
          Map<StreamsRebalanceData.HostInfo, StreamsRebalanceData.EndpointPartitions> partitionsByHost = new HashMap<>();
          data.partitionsByUserEndpoint().forEach(endpoint -> {
              List<TopicPartition> activeTopicPartitions = getTopicPartitionList(endpoint.activePartitions());
              List<TopicPartition> standbyTopicPartitions = getTopicPartitionList(endpoint.standbyPartitions());
              StreamsGroupHeartbeatResponseData.Endpoint userEndpoint = endpoint.userEndpoint();
              StreamsRebalanceData.EndpointPartitions endpointPartitions = new StreamsRebalanceData.EndpointPartitions(activeTopicPartitions, standbyTopicPartitions);
              partitionsByHost.put(new StreamsRebalanceData.HostInfo(userEndpoint.host(), userEndpoint.port()), endpointPartitions);
          });
          return partitionsByHost;
      }
  
      static List<TopicPartition> getTopicPartitionList(List<StreamsGroupHeartbeatResponseData.TopicPartition> topicPartitions) {
          return topicPartitions.stream()
                  .flatMap(partition ->
                          partition.partitions().stream().map(partitionId -> new TopicPartition(partition.topic(), partitionId)))
                  .collect(Collectors.toList());
      }
  
  }
  