public ConsumerGroupHeartbeatRequestData buildRequestData() {
            ConsumerGroupHeartbeatRequestData data = new ConsumerGroupHeartbeatRequestData();

            // GroupId - always sent
            data.setGroupId(membershipManager.groupId());

            // MemberId - always sent, empty until it has been received from the coordinator
            data.setMemberId(membershipManager.memberId());

            // MemberEpoch - always sent
            data.setMemberEpoch(membershipManager.memberEpoch());

            // InstanceId - set if present
            membershipManager.groupInstanceId().ifPresent(data::setInstanceId);

            boolean sendAllFields = membershipManager.state() == MemberState.JOINING;

            // RebalanceTimeoutMs - only sent when joining or if it has changed since the last heartbeat
            if (sendAllFields || sentFields.rebalanceTimeoutMs != rebalanceTimeoutMs) {
                data.setRebalanceTimeoutMs(rebalanceTimeoutMs);
                sentFields.rebalanceTimeoutMs = rebalanceTimeoutMs;
            }

            // SubscribedTopicNames - only sent if it has changed since the last heartbeat
            TreeSet<String> subscribedTopicNames = new TreeSet<>(this.subscriptions.subscription());
            if (sendAllFields || !subscribedTopicNames.equals(sentFields.subscribedTopicNames)) {
                data.setSubscribedTopicNames(new ArrayList<>(this.subscriptions.subscription()));
                sentFields.subscribedTopicNames = subscribedTopicNames;
            }

            // ServerAssignor - sent when joining or if it has changed since the last heartbeat
            this.membershipManager.serverAssignor().ifPresent(serverAssignor -> {
                if (sendAllFields || !serverAssignor.equals(sentFields.serverAssignor)) {
                    data.setServerAssignor(serverAssignor);
                    sentFields.serverAssignor = serverAssignor;
                }
            });

            // ClientAssignors - not supported yet

            // TopicPartitions - sent when joining or with the first heartbeat after a new assignment from
            // the server was reconciled. This is ensured by resending the topic partitions whenever the
            // local assignment, including its local epoch is changed (although the local epoch is not sent
            // in the heartbeat).
            LocalAssignment local = membershipManager.currentAssignment();
            if (sendAllFields || !local.equals(sentFields.localAssignment)) {
                List<ConsumerGroupHeartbeatRequestData.TopicPartitions> topicPartitions =
                    buildTopicPartitionsList(local.partitions);
                data.setTopicPartitions(topicPartitions);
                sentFields.localAssignment = local;
            }

            return data;
        }
Builds the heartbeat requests correctly, ensuring that all information is sent according to
the protocol, but subsequent requests do not send information which has not changed. This
is important to ensure that reconciliation completes successfully.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java