You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "m1a2st (via GitHub)" <gi...@apache.org> on 2024/07/26 13:00:35 UTC
[PR] KAFKA-17170: Add test to ensure new consumer acks reconciled assignment even if first HB with ack lost [kafka]
m1a2st opened a new pull request, #16694:
URL: https://github.com/apache/kafka/pull/16694
Jira: https://issues.apache.org/jira/browse/KAFKA-17170?
Add a unit test to the HeartbeatRequestManager
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17170: Add test to ensure new consumer acks reconciled assignment even if first HB with ack lost [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on PR #16694:
URL: https://github.com/apache/kafka/pull/16694#issuecomment-2254092871
> Hi @lianetm, I write the first version, but I consider that I miss some cases, or misunderstanding the flow, The Jira issue says that "request fails (no response) and request gets a response with an Error in it. ", but in my test don't have the error, thus I need some suggestion, Thanks for your help.
> Hi @lianetm, I write the first version, but I consider that I miss some cases, or misunderstanding the flow, The Jira issue says that "request fails (no response) and request gets a response with an Error in it. ", but in my test don't have the error, thus I need some suggestion, Thanks for your help.
How about?
```
// 2. send heartbeat1 to ack assignment tp0
time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS);
NetworkClientDelegate.PollResult result = heartbeatRequestManager.poll(time.milliseconds());
NetworkClientDelegate.UnsentRequest inflightReq = result.unsentRequests.get(0);
// 3. HB1 return error
result.unsentRequests.get(0)
.handler().onComplete(createHeartbeatResponse(inflightReq, Errors.CORRUPT_MESSAGE));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17170: Add test to ensure new consumer acks reconciled assignment even if first HB with ack lost [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #16694:
URL: https://github.com/apache/kafka/pull/16694#issuecomment-2316460431
Hello, @lianetm congratulation, I rebased it and fix the conflict, Thanks for your reminder
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17170: Add test to ensure new consumer acks reconciled assignment even if first HB with ack lost [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #16694:
URL: https://github.com/apache/kafka/pull/16694#issuecomment-2311352099
Thanks for your review, rebaed it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17170: Add test to ensure new consumer acks reconciled assignment even if first HB with ack lost [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #16694:
URL: https://github.com/apache/kafka/pull/16694#issuecomment-2316048916
Hey @m1a2st, sorry this got conflicts again before a committer could take a look, could you please rebase and solve them? (you'll have a committer available right away :) ). Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17170: Add test to ensure new consumer acks reconciled assignment even if first HB with ack lost [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16694:
URL: https://github.com/apache/kafka/pull/16694#discussion_r1722269401
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java:
##########
@@ -746,6 +746,37 @@ public void testSendingLeaveGroupHeartbeatWhenPreviousOneInFlight(final short ve
NetworkClientDelegate.PollResult pollAgain = heartbeatRequestManager.poll(time.milliseconds());
assertEquals(0, pollAgain.unsentRequests.size());
}
+
+ @ParameterizedTest
+ @ApiKeyVersionsSource(apiKey = ApiKeys.CONSUMER_GROUP_HEARTBEAT)
+ public void testConsumerAcksReconciledAssignmentWithFirstHeartBeatAckLost(final short version) {
Review Comment:
nit: would this be simpler/clearer maybe? testConsumerAcksReconciledAssignmentAfterAckLost
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java:
##########
@@ -746,6 +746,37 @@ public void testSendingLeaveGroupHeartbeatWhenPreviousOneInFlight(final short ve
NetworkClientDelegate.PollResult pollAgain = heartbeatRequestManager.poll(time.milliseconds());
assertEquals(0, pollAgain.unsentRequests.size());
}
+
+ @ParameterizedTest
+ @ApiKeyVersionsSource(apiKey = ApiKeys.CONSUMER_GROUP_HEARTBEAT)
+ public void testConsumerAcksReconciledAssignmentWithFirstHeartBeatAckLost(final short version) {
+ // 1. complete reconciliation
+ createHeartbeatStatAndRequestManager();
+ String topic = "topic1";
+ int exceededTimeMs = 5;
+ Set<String> set = Collections.singleton(topic);
+ when(subscriptions.subscription()).thenReturn(set);
+ subscriptions.subscribe(set, Optional.empty());
+ mockReconcilingMemberData();
+ // 2. send heartbeat1 to ack assignment tp0
+ time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS);
+ NetworkClientDelegate.PollResult result = heartbeatRequestManager.poll(time.milliseconds());
+ // 3. HB1 times out
+ result.unsentRequests.get(0)
+ .handler()
+ .onFailure(time.milliseconds(), new TimeoutException("timeout"));
+ // 4. heartbeat request manager resets the sentFields to null HeartbeatState.reset()
+ time.sleep(DEFAULT_MAX_POLL_INTERVAL_MS + exceededTimeMs);
+ assertHeartbeat(heartbeatRequestManager, DEFAULT_HEARTBEAT_INTERVAL_MS);
+ verify(heartbeatRequestState).reset();
+ // 5. following HB will include tp0 (and act as ack), tp0 != null
+ result = heartbeatRequestManager.poll(time.milliseconds());
+ NetworkClientDelegate.UnsentRequest request = result.unsentRequests.get(0);
+ ConsumerGroupHeartbeatRequest heartbeatRequest =
+ (ConsumerGroupHeartbeatRequest) request.requestBuilder().build(version);
+
+ assertEquals(Collections.singletonList(topic), heartbeatRequest.data().subscribedTopicNames());
Review Comment:
In this situation (ack lost), we expect that the member should resend the partitions, not only the topic names. So should we assert it does? (we probably need to pass the partitions into the `mockReconcilingMemberData`, to be returned in the currentAssignment on ln 919, and then assert that the same partitions are indeed in the `heartbeatRequest.data().topicPartitions()`
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java:
##########
@@ -746,6 +746,37 @@ public void testSendingLeaveGroupHeartbeatWhenPreviousOneInFlight(final short ve
NetworkClientDelegate.PollResult pollAgain = heartbeatRequestManager.poll(time.milliseconds());
assertEquals(0, pollAgain.unsentRequests.size());
}
+
+ @ParameterizedTest
+ @ApiKeyVersionsSource(apiKey = ApiKeys.CONSUMER_GROUP_HEARTBEAT)
+ public void testConsumerAcksReconciledAssignmentWithFirstHeartBeatAckLost(final short version) {
+ // 1. complete reconciliation
+ createHeartbeatStatAndRequestManager();
Review Comment:
since we're here, let's please fix the typo in createHeartbeat**State**AndRequestManager
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java:
##########
@@ -746,6 +746,37 @@ public void testSendingLeaveGroupHeartbeatWhenPreviousOneInFlight(final short ve
NetworkClientDelegate.PollResult pollAgain = heartbeatRequestManager.poll(time.milliseconds());
assertEquals(0, pollAgain.unsentRequests.size());
}
+
+ @ParameterizedTest
+ @ApiKeyVersionsSource(apiKey = ApiKeys.CONSUMER_GROUP_HEARTBEAT)
+ public void testConsumerAcksReconciledAssignmentWithFirstHeartBeatAckLost(final short version) {
+ // 1. complete reconciliation
+ createHeartbeatStatAndRequestManager();
+ String topic = "topic1";
+ int exceededTimeMs = 5;
+ Set<String> set = Collections.singleton(topic);
+ when(subscriptions.subscription()).thenReturn(set);
+ subscriptions.subscribe(set, Optional.empty());
+ mockReconcilingMemberData();
+ // 2. send heartbeat1 to ack assignment tp0
+ time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS);
+ NetworkClientDelegate.PollResult result = heartbeatRequestManager.poll(time.milliseconds());
+ // 3. HB1 times out
+ result.unsentRequests.get(0)
+ .handler()
+ .onFailure(time.milliseconds(), new TimeoutException("timeout"));
+ // 4. heartbeat request manager resets the sentFields to null HeartbeatState.reset()
+ time.sleep(DEFAULT_MAX_POLL_INTERVAL_MS + exceededTimeMs);
Review Comment:
I would expect that just sleeping the interval would be enough, so we could maybe simplify the test and remove the `exceededTimeMs` from here and the var itself?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17170: Add test to ensure new consumer acks reconciled assignment even if first HB with ack lost [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm merged PR #16694:
URL: https://github.com/apache/kafka/pull/16694
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17170: Add test to ensure new consumer acks reconciled assignment even if first HB with ack lost [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #16694:
URL: https://github.com/apache/kafka/pull/16694#issuecomment-2253724815
Hi @lianetm, I write the first version, but I consider that I miss some case, or misunderstanding the flow, The Jira issue says that "request fails (no response) and request gets a response with an Error in it. ", but in my test don't have the error, thus I need some suggestion, Thanks for your help.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17170: Add test to ensure new consumer acks reconciled assignment even if first HB with ack lost [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #16694:
URL: https://github.com/apache/kafka/pull/16694#discussion_r1735417379
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java:
##########
@@ -746,6 +746,51 @@ public void testSendingLeaveGroupHeartbeatWhenPreviousOneInFlight(final short ve
NetworkClientDelegate.PollResult pollAgain = heartbeatRequestManager.poll(time.milliseconds());
assertEquals(0, pollAgain.unsentRequests.size());
}
+
+ @ParameterizedTest
+ @ApiKeyVersionsSource(apiKey = ApiKeys.CONSUMER_GROUP_HEARTBEAT)
+ public void testConsumerAcksReconciledAssignmentAfterAckLost(final short version) {
+ String topic = "topic1";
+ Set<String> topics = Collections.singleton(topic);
+ Uuid topicId = Uuid.randomUuid();
+ int partition = 0;
+ Map<Uuid, SortedSet<Integer>> testAssignment = Collections.singletonMap(
+ topicId, mkSortedSet(partition)
+ );
+
+ // complete reconciliation
+ createHeartbeatStateAndRequestManager();
+ when(subscriptions.subscription()).thenReturn(topics);
+ subscriptions.subscribe(topics, Optional.empty());
+ mockReconcilingMemberData(testAssignment);
+
+ // send heartbeat1 to ack assignment tp0
+ time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS);
+ NetworkClientDelegate.PollResult result = heartbeatRequestManager.poll(time.milliseconds());
+
+ // HB1 times out
+ result.unsentRequests.get(0)
Review Comment:
It is good idea, I will address it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17170: Add test to ensure new consumer acks reconciled assignment even if first HB with ack lost [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #16694:
URL: https://github.com/apache/kafka/pull/16694#discussion_r1735413135
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java:
##########
@@ -746,6 +746,51 @@ public void testSendingLeaveGroupHeartbeatWhenPreviousOneInFlight(final short ve
NetworkClientDelegate.PollResult pollAgain = heartbeatRequestManager.poll(time.milliseconds());
assertEquals(0, pollAgain.unsentRequests.size());
}
+
+ @ParameterizedTest
+ @ApiKeyVersionsSource(apiKey = ApiKeys.CONSUMER_GROUP_HEARTBEAT)
+ public void testConsumerAcksReconciledAssignmentAfterAckLost(final short version) {
+ String topic = "topic1";
+ Set<String> topics = Collections.singleton(topic);
+ Uuid topicId = Uuid.randomUuid();
+ int partition = 0;
+ Map<Uuid, SortedSet<Integer>> testAssignment = Collections.singletonMap(
+ topicId, mkSortedSet(partition)
+ );
+
+ // complete reconciliation
+ createHeartbeatStateAndRequestManager();
+ when(subscriptions.subscription()).thenReturn(topics);
+ subscriptions.subscribe(topics, Optional.empty());
+ mockReconcilingMemberData(testAssignment);
+
+ // send heartbeat1 to ack assignment tp0
+ time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS);
+ NetworkClientDelegate.PollResult result = heartbeatRequestManager.poll(time.milliseconds());
+
+ // HB1 times out
+ result.unsentRequests.get(0)
Review Comment:
It would be ideal to assert that the number of requests is >= 1 before getting the first element, only because if we ever have a regression somewhere and it returns an empty request list, the `IndexOutOfBoundsException` (or whatever) won't be very helpful in tracking down the problem.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17170: Add test to ensure new consumer acks reconciled assignment even if first HB with ack lost [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #16694:
URL: https://github.com/apache/kafka/pull/16694#issuecomment-2254285554
@TaiJuWu, Thanks for your comment, I will take a look your refer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17170: Add test to ensure new consumer acks reconciled assignment even if first HB with ack lost [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #16694:
URL: https://github.com/apache/kafka/pull/16694#issuecomment-2299427861
Hello @lianetm, Thanks for your comments, I have been addressed all the comments, PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org