showuon commented on a change in pull request #11971:
URL: https://github.com/apache/kafka/pull/11971#discussion_r839220945
##########
File path:
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
##########
@@ -2819,6 +2851,39 @@ public void
testEnforceRebalanceTriggersRebalanceOnNextPoll() {
assertEquals(countingRebalanceListener.revokedCount, 1);
}
+ @Test
+ public void testEnforceRebalanceReason() {
+ Time time = new MockTime(1L);
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ KafkaConsumer<String, String> consumer = newConsumer(time, client,
subscription, metadata, assignor, true, groupInstanceId);
+ MockRebalanceListener countingRebalanceListener = new
MockRebalanceListener();
+ initMetadata(client, Utils.mkMap(Utils.mkEntry(topic, 1),
Utils.mkEntry(topic2, 1), Utils.mkEntry(topic3, 1)));
+
+ consumer.subscribe(Arrays.asList(topic, topic2),
countingRebalanceListener);
+ Node node = metadata.fetch().nodes().get(0);
+ prepareRebalance(client, node, assignor, Arrays.asList(tp0, t2p0),
null);
+
+ // a first rebalance to get the assignment, we need two poll calls
since we need two round trips to finish join / sync-group
+ consumer.poll(Duration.ZERO);
+ consumer.poll(Duration.ZERO);
Review comment:
I think what we want is to wait for rebalance complete. we can
explicitly wait like this:
```java
TestUtils.waitForCondition(() -> {
ConsumerRecords<String, String> recs =
consumer.poll(Duration.ofMillis(100L));
return consumer.assignment().equals(Utils.mkSet(tp0, t2p0));
});
```
WDYT?
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
