sjhajharia commented on code in PR #22714:
URL: https://github.com/apache/kafka/pull/22714#discussion_r3504092086
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java:
##########
@@ -2033,4 +2034,61 @@ public void
testDescribeShareGroupOffsetsForEmptySharePartition() {
throw new RuntimeException(e);
}
}
+
+ @ClusterTest
+ public void testAddedPartitionsOfKnownTopicStartAtOffsetZero() throws
InterruptedException, ExecutionException {
+ String groupId = "expand-group";
+ String topic = "expand-topic";
+ createTopic(topic, 1, 1);
+
+ alterShareAutoOffsetReset(groupId, "latest");
+
+ try (Producer<byte[], byte[]> producer = createProducer();
+ Admin admin = createAdminClient();
+ ShareConsumer<byte[], byte[]> shareConsumer =
createShareConsumer(groupId)) {
+
+ // Produce 10 records to partition 0 before anyone subscribes.
+ for (int i = 0; i < 10; i++) {
+ producer.send(new ProducerRecord<>(topic, 0, null, ("p0-key-"
+ i).getBytes(), ("p0-value-" + i).getBytes()));
+ }
+ producer.flush();
+
+ // The consumer joins and polls: with "latest" it starts partition
0 at the log end, so it receives
+ // none of the pre-existing records. This also makes the group
aware of (initializes) the topic.
+ shareConsumer.subscribe(Set.of(topic));
+ int received = 0;
+ long deadline = System.currentTimeMillis() + 10000L;
+ while (System.currentTimeMillis() < deadline) {
+ received +=
shareConsumer.poll(Duration.ofMillis(1000)).count();
+ }
+ assertEquals(0, received, "latest reset should skip records
produced before the topic was known");
+
+ // Expand the topic and produce 10 records to the brand-new
partition 1, each with a distinct,
+ // known value so we can assert all of them are delivered (none
skipped by the latest reset).
+ admin.createPartitions(Map.of(topic,
NewPartitions.increaseTo(2))).all().get();
+ cluster.waitTopicCreation(topic, 2);
+
+ List<String> expectedValues = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ String value = "p1-value-" + i;
+ expectedValues.add(value);
+ producer.send(new ProducerRecord<>(topic, 1, null, ("p1-key-"
+ i).getBytes(), value.getBytes()));
+ }
+ producer.flush();
+
+ // Partition 1 belongs to an already-known topic, so it must be
initialized at offset 0 and deliver
+ // every record produced above, in order — even though it was
created under a "latest" reset.
+ List<String> receivedValues = new ArrayList<>();
+ waitForCondition(() -> {
+ ConsumerRecords<byte[], byte[]> records =
shareConsumer.poll(Duration.ofMillis(1000));
+ for (ConsumerRecord<byte[], byte[]> record : records) {
+ if (record.partition() == 1) {
+ receivedValues.add(new String(record.value()));
+ }
+ }
+ return receivedValues.size() >= expectedValues.size();
+ }, 30000L, 500L, () -> "did not receive all records from the newly
added partition");
+ assertEquals(expectedValues, receivedValues);
Review Comment:
Thanks for the review! Yes, I noticed that given the at-least-once semantics
of Share Group, we may have flakes here. I have updated the same to assert on
the size of Set.
--
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]
