Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java (Java)

Code before:
            this.rebalanceTimeoutMs = rebalanceTimeoutMs;
            this.sentFields = new SentFields();
        }


        public void reset() {
            sentFields.reset();
        }

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

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

            // MemberId - always sent, it will be generated at Consumer startup.
            data.setMemberId(membershipManager.memberId());

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

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

            // RackId - set if present
            membershipManager.rackId().ifPresent(data::setRackId);

            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;
            }

            // SubscribedTopicRegex - only sent if it has changed since the last heartbeat.
            // Send empty string to indicate that a subscribed pattern needs to be removed.
            SubscriptionPattern pattern = subscriptions.subscriptionPattern();
            boolean patternUpdated = !Objects.equals(pattern, sentFields.pattern);
            if ((sendAllFields && pattern != null) || patternUpdated) {
                data.setSubscribedTopicRegex((pattern != null) ? pattern.pattern() : "");
                sentFields.pattern = pattern;
            }


Review comment (suggestion):
I think that it should only be set if it did not change since the last request or if we send a full request.

Code after:
            this.rebalanceTimeoutMs = rebalanceTimeoutMs;
            this.sentFields = new SentFields();
        }


        public void reset() {
            sentFields.reset();
        }

        @SuppressWarnings("NPathComplexity")
        public ConsumerGroupHeartbeatRequestData buildRequestData() {
            ConsumerGroupHeartbeatRequestData data = new ConsumerGroupHeartbeatRequestData();

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

            // MemberId - always sent, it will be generated at Consumer startup.
            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;
            }

            // SubscribedTopicRegex - only sent if it has changed since the last heartbeat.
            // Send empty string to indicate that a subscribed pattern needs to be removed.
            SubscriptionPattern pattern = subscriptions.subscriptionPattern();
            boolean patternUpdated = !Objects.equals(pattern, sentFields.pattern);
            if ((sendAllFields && pattern != null) || patternUpdated) {
                data.setSubscribedTopicRegex((pattern != null) ? pattern.pattern() : "");
                sentFields.pattern = pattern;
            }

            // ServerAssignor - sent when joining or if it has changed since the last heartbeat
            this.membershipManager.serverAssignor().ifPresent(serverAssignor -> {