showuon commented on code in PR #12349:
URL: https://github.com/apache/kafka/pull/12349#discussion_r924022486
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java:
##########
@@ -1299,6 +1377,178 @@ public void
testForceMetadataDeleteForPatternSubscriptionDuringRebalance() {
}
}
+ @Test
+ public void testOnJoinPrepareWithOffsetCommitShouldSuccessAfterRetry() {
+ rebalanceConfig = buildRebalanceConfig(Optional.of("group-id"));
+ ConsumerCoordinator coordinator = buildCoordinator(rebalanceConfig,
+ new Metrics(),
+ assignors,
+ true,
+ subscriptions);
+ client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
+ coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));
+ subscriptions.subscribe(singleton(topic1), rebalanceListener);
+ client.prepareResponse(joinGroupFollowerResponse(1, consumerId,
"leader", Errors.NONE));
+ client.prepareResponse(syncGroupResponse(singletonList(t1p),
Errors.NONE));
+ coordinator.joinGroupIfNeeded(time.timer(Long.MAX_VALUE));
Review Comment:
Also, since we created a new coordinator, we should close it at the end of
test. Otherwise, it'll break other tests. If you agree with my above
suggestion, we can do like this to close it:
```java
try (ConsumerCoordinator coordinator = prepareCoordinatorForCloseTest(...)) {
}
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
