KAFKA-16528: Client HB timing fix #15698

lianetm · 2024-04-11T13:23:10Z

Fix for resetting HB timer when the request is sent, rather than when a response is received. This ensures a more accurate timing of the HB, so that a member always sends HB on the interval (not in the interval + any delay in receiving the response).
This change, along with the logic already in place for checking in-flights, ensures that if the interval expires but there is a HB in-flight, the next HB is only send after the response for the in-flight is received, without waiting for another full interval. This is btw consistent with the timer reset & inflight behaviour for the auto-commit interval.

lianetm · 2024-04-11T19:43:43Z

Hey @cadonna , could you take a look at this small fix when you have a chance? Thanks!

cadonna

Thanks for the PR, @lianetm !

Here my feedback.

cadonna · 2024-04-12T12:44:27Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java

+        // Reset timer when sending the request, to make sure that, if waiting for the interval,
+        // we don't include the response time (which may introduce delay)


Do we really need this comment?
Additionally, I could not find a verification of this call in unit tests. Since you added a comment it seems to be important enough for a verification.

You're right, I don't think it's bringing anything not clear with the func name and action themselves. Removed.
This is covered in the new test I added here. I just included now an assert message along the lines of this comment to make it clearer in the test.

...s/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java

cadonna · 2024-04-12T13:13:08Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java

+            // Expire timer to allow sending HB after a failure. After a failure, a next HB may be
+            // needed with backoff (ex. errors that lead to retries, like coordinator load error),
+            // or immediately (ex. errors that lead to rejoining, like fencing errors).
+            heartbeatTimer.update(heartbeatTimer.currentTimeMs() + heartbeatTimer.remainingMs());


What about adding a method to Timer that expires the timer without updating it with a time point in the future. Alternatively, I think you could reset the Timer to 0 with heartbeatTimer.reset(0).
Do we need a verification in the unit tests for this?

Done, good point. I changed it to reset to 0, that shows the intention of not having an interval to wait for, which is what we want on these failure scenarios.

Your change to heartbeatTimer.reset(0) is not totally equivalent to heartbeatTimer.update(heartbeatTimer.currentTimeMs() + heartbeatTimer.remainingMs()); because in the former the heartbeatTimer is not expired until heartbeatTimer.update() is called, whereas in the latter the heartbeat is expired after the call, but I think in this specific case it does not matter. Is my assumption correct?

Sorry, my previous comment is not correct. The heartbeatTimer is expired when you call heartbeatTimer.reset(0).

...s/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java

lianetm · 2024-04-15T13:38:44Z

Hey @cadonna, thanks a lot for your feedback! All comments addressed.

kirktrue

Thanks for this fix, @lianetm!

One question: I noticed that the heartbeat timer is reset on errors. In testHeartbeatNotSentIfAnotherOneInFlight, it mimics a delayed, but successful response. Does it make sense to add another unit test where instead the delayed response has an error? My understanding from this patch is that we could don't have to sleep at all and the next call to poll() should return the next request.

Apologies if/that I'm confused 😄

Thanks!

kirktrue · 2024-04-16T22:09:52Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java

-                logger.error("GroupHeartbeatRequest failed due to the instance id {} was not released: {}",
+                logger.error("GroupHeartbeatRequest failed due to unreleased instance id {}: {}",


QQ: are these logging changes of the ‘I'll just clean this up as long as I'm in here?’ variety, or dow it have some bearing on the correctness of the logs?

both. The change for the default log is needed: it was not including the errorMessage, and that makes it hard to know what happened when you get errors like INVALID_REQUEST (I personally got it and lost time investigating, so fixed it). The other log changes are just improvements because I was already there.

kirktrue · 2024-04-16T22:11:29Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java

+            // Reset timer to allow sending HB after a failure without waiting for the interval.
+            // After a failure, a next HB may be needed with backoff (ex. errors that lead to
+            // retries, like coordinator load error), or immediately (ex. errors that lead to
+            // rejoining, like fencing errors).
+            heartbeatTimer.reset(0);
+            super.onFailedAttempt(currentTimeMs);


Does the decision to reset the heartbeat timer depend on what type of error is received?

no, it doesn't. The timer is only to indicate that an interval should be respected. In cases of failures, we don't want to follow the interval (so we reset timer to 0). Each error will :

send a next HB based on other conditions (ex. as soon as the coordinator is discovered, when releasing assignment finishes after getting fenced)

not send a next HB at all (fatal errors)

kirktrue · 2024-04-16T22:12:19Z

...s/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java

+
+        result = heartbeatRequestManager.poll(time.milliseconds());
+        assertEquals(0, result.unsentRequests.size(), "No heartbeat should be sent while a " +
+            "previous on in-flight");


Super nit-picky, sorry 😞

Suggested change

"previous on in-flight");

"previous one in-flight");

Fixed (not nit-picky at all, ugly typo, good catch!)

lianetm · 2024-04-17T20:44:42Z

Thanks for the comments @kirktrue, all addressed.

Regarding your comment regarding tests here, we have that covered with the existing testHeartbeatResponseOnErrorHandling. That one is validating that we get the right nextHeartbeatMs time (which considers the timer), for each specific error type. Is that what you were looking for? Thanks!

lianetm · 2024-04-18T15:33:35Z

Hey @cadonna, could you take a look when you have chance? Thanks!

kirktrue

Thanks for this patch, @lianetm!

I appreciate the educational responses to my naive questions 😄

cadonna

Thanks for the updates, @lianetm !

I wondering why no unit test is failing after your changes. That tells me that unit tests are missing that verify that the interval is reset correctly.

I am also a bit concerned that I could not find unit tests that verify that HeartbeatRequestState#canSendRequest() and HeartbeatRequestState#nextHeartbeatMs() work consistently. With this I mean that when the former returns true the latter returns 0. Or that when the former returns false the latter returns a time > 0.

lianetm · 2024-04-22T14:45:20Z

Hi @cadonna, thanks for the comments!

the unit test I added initially fails on this assertion without the changes on this PR, I guess that's what you were looking for with your first comment?
agreed, we did not have test coverage to make sure that the canSendRequest and nextHeartbeat move along consistently. I added a func to validate them together, included it in the existing test that covers responses with errors, and added a new test to include it in the successful path too.

cadonna

Thanks for the updates, @lianetm !

Here my feedback.

cadonna · 2024-04-25T09:17:43Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java

        @Override
        public boolean canSendRequest(final long currentTimeMs) {
            update(currentTimeMs);
            return heartbeatTimer.isExpired() && super.canSendRequest(currentTimeMs);
        }

-        public long nextHeartbeatMs(final long currentTimeMs) {
+        public long timeToNextHeartbeatMs(final long currentTimeMs) {
            if (heartbeatTimer.remainingMs() == 0) {


Sorry if I comment on code outside the PR. Isn't this the same as heartbeatTimer.isExpired()? If yes, could we please change this to make the the code more readable?

They achieve the same here, and totally agree that isExpired is more readable, fixed. (Sensible "since we're here..." to me too btw)

cadonna · 2024-04-25T09:41:56Z

...s/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java

+        assertNextHeartbeatTiming(DEFAULT_HEARTBEAT_INTERVAL_MS);
+
+        result = heartbeatRequestManager.poll(time.milliseconds());
+        assertEquals(1, result.unsentRequests.size(), "A heartbeat should be sent when interval expires");


Don't you need a verification here that ensures that the heartbeat timer was reset after the poll?

Sure! I missed it. Added.

cadonna · 2024-04-25T09:45:07Z

...s/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java

+        assertEquals(1, result.unsentRequests.size(), "A heartbeat should be sent when interval expires");
+        NetworkClientDelegate.UnsentRequest inflightReq = result.unsentRequests.get(0);
+        inflightReq.handler().onComplete(createHeartbeatResponse(inflightReq, Errors.NONE));
+        assertNextHeartbeatTiming(DEFAULT_HEARTBEAT_INTERVAL_MS);


How do you know whether the heartbeat timer was reset in makeHeartbeatRequest() or in onResponse() if you verify the reset after you complete the future of the request?
I would verify the timer reset after the poll, progress the time a bit (less then the heartbeat interval), and then verify here that the time to the next heartbeat is the heartbeat interval minus the amount of time I progressed the time after the poll.

This relates to your comment above. You're right that we did not have the check to ensure the reset happened on the sent and not on the response, so I added above the check for the timeToNextHeartbeatMs that would fail if the timer is not reset on the send, with a specific message for it. That check covers it, but still I also added the steps for advance the timer just a bit, check that no HB is sent and that the time is updated with the difference, as you suggested. All done.

lianetm · 2024-04-25T14:15:07Z

Thanks for the helpful comments @cadonna , all addressed.

cadonna

Thanks for the updates and considering my feedback @lianetm !

LGTM!

lianetm added 4 commits April 11, 2024 09:15

Reset timer on hb sent

df4a87f

Add inflight test

f014f8d

Improve logs

1684eec

Update test

ad0f137

lianetm changed the title ~~Client HB timing fixes~~ KAFKA-16528: Client HB timing fix Apr 11, 2024

lianetm marked this pull request as ready for review April 11, 2024 19:39

cadonna reviewed Apr 12, 2024

View reviewed changes

lianetm added 3 commits April 15, 2024 09:16

Remove comment

4b4df70

Adding assert message & fix test comment

bbd0344

Change for expiring interval on error

0b74f57

lianetm added 2 commits April 15, 2024 09:40

Update comment

fe483ff

Merge branch 'trunk' into hb-timer-fix

568bc4a

kirktrue reviewed Apr 16, 2024

View reviewed changes

Fix typo

a47584d

kirktrue approved these changes Apr 18, 2024

View reviewed changes

cadonna reviewed Apr 22, 2024

View reviewed changes

lianetm added 2 commits April 22, 2024 10:11

Improve func name and add comment

9463267

Adding tests

d242c44

cadonna reviewed Apr 25, 2024

View reviewed changes

lianetm added 2 commits April 25, 2024 09:40

Use isExpired check

bec3f62

Extend test for HB timing on successful path

5354ebd

Merge branch 'trunk' into hb-timer-fix

569f574

cadonna approved these changes Apr 26, 2024

View reviewed changes

cadonna merged commit 82844c0 into apache:trunk Apr 26, 2024
1 check failed

chickenchickenlove mentioned this pull request May 11, 2024

GH-3145 : Add test for next generation consumer-group rebalance protocol spring-projects/spring-kafka#3237

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-16528: Client HB timing fix #15698

KAFKA-16528: Client HB timing fix #15698

lianetm commented Apr 11, 2024 •

edited

lianetm commented Apr 11, 2024

cadonna left a comment

cadonna Apr 12, 2024

lianetm Apr 15, 2024 •

edited

cadonna Apr 12, 2024

lianetm Apr 15, 2024

cadonna Apr 22, 2024

cadonna Apr 22, 2024

lianetm commented Apr 15, 2024

kirktrue left a comment

kirktrue Apr 16, 2024

lianetm Apr 17, 2024

kirktrue Apr 16, 2024

lianetm Apr 17, 2024

kirktrue Apr 16, 2024

lianetm Apr 17, 2024

lianetm commented Apr 17, 2024

lianetm commented Apr 18, 2024

kirktrue left a comment

cadonna left a comment

lianetm commented Apr 22, 2024

cadonna left a comment

cadonna Apr 25, 2024

lianetm Apr 25, 2024

cadonna Apr 25, 2024

lianetm Apr 25, 2024

cadonna Apr 25, 2024 •

edited

lianetm Apr 25, 2024

lianetm commented Apr 25, 2024

cadonna left a comment

		// Reset timer when sending the request, to make sure that, if waiting for the interval,
		// we don't include the response time (which may introduce delay)

		logger.error("GroupHeartbeatRequest failed due to the instance id {} was not released: {}",
		logger.error("GroupHeartbeatRequest failed due to unreleased instance id {}: {}",

KAFKA-16528: Client HB timing fix #15698

KAFKA-16528: Client HB timing fix #15698

Conversation

lianetm commented Apr 11, 2024 • edited

lianetm commented Apr 11, 2024

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lianetm Apr 15, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lianetm commented Apr 15, 2024

kirktrue left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lianetm commented Apr 17, 2024

lianetm commented Apr 18, 2024

kirktrue left a comment

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment

lianetm commented Apr 22, 2024

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna Apr 25, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lianetm commented Apr 25, 2024

cadonna left a comment

Choose a reason for hiding this comment

lianetm commented Apr 11, 2024 •

edited

lianetm Apr 15, 2024 •

edited

cadonna Apr 25, 2024 •

edited