smjn commented on code in PR #23955:
URL: https://github.com/apache/kafka/pull/23955#discussion_r4376780213
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java:
##########
@@ -1626,29 +1625,26 @@ public void
testSubscriptionAndPollFollowedByTopicDeletion() throws InterruptedE
ProducerRecord<byte[], byte[]> recordTopic2 = new
ProducerRecord<>(topic2, 0, null, "key".getBytes(), "value".getBytes());
// Consumer subscribes to the topics -> bar and baz.
- shareConsumer.subscribe(Arrays.asList(topic1, topic2));
+ shareConsumer.subscribe(List.of(topic1, topic2));
producer.send(recordTopic1).get();
- TestUtils.waitForCondition(() ->
shareConsumer.poll(Duration.ofMillis(2000)).count() == 1,
- DEFAULT_MAX_WAIT_MS, 100L, () -> "incorrect number of
records");
+ ConsumerRecords<byte[], byte[]> records =
waitedPoll(shareConsumer, 2000L, 1);
+ assertEquals(topic1, records.iterator().next().topic());
producer.send(recordTopic2).get();
- TestUtils.waitForCondition(() ->
shareConsumer.poll(Duration.ofMillis(2000)).count() == 1,
- DEFAULT_MAX_WAIT_MS, 100L, () -> "incorrect number of
records");
+ records = waitedPoll(shareConsumer, 2000L, 1);
+ assertEquals(topic2, records.iterator().next().topic());
- // Topic bar is deleted, hence poll should not give any results.
+ // Topic bar is deleted, so its partition should be revoked and
the member should end up assigned
+ // only the partition of topic baz. The subscription itself is
unchanged.
deleteTopic(topic1);
- ConsumerRecords<byte[], byte[]> records =
shareConsumer.poll(Duration.ofMillis(500));
- assertEquals(0, records.count());
+ waitForExactAssignment("group1", Set.of(new TopicPartition(topic2,
0)));
Review Comment:
nice one
--
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]
