FrankChen021 commented on code in PR #19817:
URL: https://github.com/apache/druid/pull/19817#discussion_r3686804498
##########
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/KafkaResourceTest.java:
##########
@@ -58,8 +59,19 @@ public void testKafka()
// Test topic creation
final String topicName = "test-topic";
- resource.createTopicWithPartitions(topicName, 3);
+ resource.createTopicWithPartitions(topicName, 2);
assertEquals(Set.of(topicName), resource.listTopics());
+
+ // Verify that records can be published immediately after adding
partitions.
+ resource.increasePartitionsInTopic(topicName, 4);
+ resource.publishRecordsToTopicWithoutTransaction(
+ topicName,
+ Collections.nCopies(1_000, new byte[]{1})
+ );
+ final Map<String, Long> partitionOffsets =
resource.getPartitionOffsets(topicName);
+ assertEquals(4, partitionOffsets.size());
+ assertEquals(1_000,
partitionOffsets.values().stream().mapToLong(Long::longValue).sum());
Review Comment:
Addressed in commit 3fc8dfab92. The regression now publishes explicit
records to every initial partition and to partitions 0 through 3 immediately
after expansion, then asserts each partition end offset individually.
KafkaResourceTest and both partition-change fault-tolerance parameterizations
pass.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
