smjn commented on code in PR #22714:
URL: https://github.com/apache/kafka/pull/22714#discussion_r3503995560
##########
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:
can this cause flakiness if there are duplicate records?
--
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]
