KAFKA-18736: Decide when a heartbeat should be sent

[cadonna]

This commit adds the conditions to decide when a Streams group heartbeat should be sent.
A heartbeat should be sent when:

• the group coordinator is available
• the member is part of the Streams group or wants to join it
• the heartbeat interval expired or the member is leaving the group or acknowledging the assginment

This commit does not implement:

• not sending fields that did not change
• handling errors

Reviewers: Zheguang Zhao zheguang.zhao@alumni.brown.edu, Lucas Brutschy lbrutschy@confluent.io

[cadonna]

Call for review: @aliehsaeedii

[Copilot]

PR Overview

This pull request refactors the heartbeat request management for Kafka Streams by introducing new conditions and timer logic to decide when a heartbeat should be sent. Key changes include:

• Adding a poll timer and new logic in the heartbeat polling method to handle poll timeouts and skip conditions.
• Renaming and updating methods in StreamsMembershipManager (from shouldHeartbeatNow to shouldNotWaitForHeartbeatInterval) along with corresponding test updates.
• Adding detailed JavaDoc comments and new tests to cover various member states and heartbeat scenarios.

Reviewed Changes

File	Description
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java	Introduces poll timer logic, new conditional branches for sending heartbeats, and improved error propagation via BackgroundEventHandler.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java	Updates and new tests to validate heartbeat request behavior under different member states and timer conditions.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java	Renames method ensuring consistency in heartbeat interval decision making.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManagerTest.java	Adjusts tests to match the updated method naming and expected behavior.

Copilot reviewed 4 out of 4 changed files in this pull request and generated no comments.

Comments suppressed due to low confidence (2)

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:352

• [nitpick] Consider adding parentheses in the compound condition within shouldHeartbeatBeforeIntervalExpires() to improve clarity of the evaluation order, e.g. by grouping the JOINING and ACKNOWLEDGING checks together before applying the requestInFlight() negation.

private boolean shouldHeartbeatBeforeIntervalExpires() {

clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java:650

• The heartbeat interval value is cast from long to int; ensure that the interval value never exceeds the limits of an int to prevent potential overflow issues in production.

.setHeartbeatIntervalMs((int) RECEIVED_HEARTBEAT_INTERVAL_MS)

[zheguang]

🤔 wondering if the closing parenthesis is off by one clause? By reading the comment above, it seems this intended instead:

||
            (membershipManager.state() == MemberState.JOINING || membershipManager.state() == MemberState.ACKNOWLEDGING
                && !heartbeatRequestState.requestInFlight());```

[cadonna]

@zheguang Thanks for your comment!

The comment says:

the member is joining the group or acknowledging the assignment and for both cases there is no heartbeat request in flight.

That means, the expression should be true for

membershipManager.state() == MemberState.JOINING && !heartbeatRequestState.requestInFlight()

OR

membershipManager.state() == MemberState.ACKNOWLEDGING && !heartbeatRequestState.requestInFlight()

The distributive laws of boolean algrebra say:

(B⋅A)+(C⋅A) = (B+C)⋅A 

If we define A to be !heartbeatRequestState.requestInFlight(), B to be membershipManager.state() == MemberState.JOINING && !heartbeatRequestState.requestInFlight(), and C to be membershipManager.state() == MemberState.ACKNOWLEDGING && !heartbeatRequestState.requestInFlight(), we can rewrite the above to

(membershipManager.state() == MemberState.JOINING || membershipManager.state() == MemberState.ACKNOWLEDGING) 
&& !heartbeatRequestState.requestInFlight()

If we applied your suggestion, the expression would be true if

membershipManager.state() == MemberState.JOINING

OR

membershipManager.state() == MemberState.ACKNOWLEDGING && !heartbeatRequestState.requestInFlight()

So, the member state needs only to be MemberState.JOINING to send a heartbeat. It would not check for a heartbeat in flight.

Do you agree?

[lucasbru]

THanks for the PR, @cadonna ! Here are my comments

[lucasbru]

These are not separate situations in which a heartbeat must be sent. Maybe you mean "when all of the following conditions apply" or soemthign?

[lucasbru]

something seems to be missing after @return.

[lucasbru]

Wow, this addCompletionCallback is complex, and also seems quite unncessary. Can I not just do a handleResponse(makeHeartbeatRequest(currentTimeMs)) or -even better in my eyes - implement a method for the lambda inside of handleResponse only, and then do a makeHeartbeatRequest(currentTimeMs).whenComplete(handleResponse).

[cadonna]

Nice idea! Thanks!
Let me know how you like it now.

[lucasbru]

yeah, that's better!

[lucasbru]

LGTM, thanks!