FrankChen021 commented on code in PR #19824:
URL: https://github.com/apache/druid/pull/19824#discussion_r3686786276
##########
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/KafkaResource.java:
##########
@@ -134,6 +135,15 @@ public void createTopicWithPartitions(String topicName,
int numPartitions)
admin.createTopics(
List.of(new NewTopic(topicName, numPartitions, (short) 1))
).all().get();
+
+ // createTopics() may complete before the partition leaders are ready to
+ // handle requests. Verify every partition through its leader before
+ // allowing callers to start a supervisor or publish records.
+ final Map<TopicPartition, OffsetSpec> partitionOffsets = new HashMap<>();
+ for (int partition = 0; partition < numPartitions; partition++) {
+ partitionOffsets.put(new TopicPartition(topicName, partition),
OffsetSpec.latest());
+ }
+ admin.listOffsets(partitionOffsets).all().get();
Review Comment:
This PR was closed as superseded by #19817. The follow-up on #19817 in
commit 3b49ef3f3f adds a bounded five-attempt retry for listOffsets and retries
only Kafka RetriableException failures, including wrapped causes. The focused
KafkaResourceTest passes.
##########
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/KafkaResourceTest.java:
##########
@@ -60,6 +61,16 @@ public void testKafka()
final String topicName = "test-topic";
resource.createTopicWithPartitions(topicName, 3);
assertEquals(Set.of(topicName), resource.listTopics());
+
+ // Verify that callers can publish immediately after topic creation.
+ resource.publishRecordsToTopicWithoutTransaction(
+ topicName,
+ Collections.nCopies(1_000, new byte[]{1})
+ );
+ final Map<String, Long> partitionOffsets =
resource.getPartitionOffsets(topicName);
+ assertEquals(3, partitionOffsets.size());
+ assertEquals(1_000,
partitionOffsets.values().stream().mapToLong(Long::longValue).sum());
Review Comment:
This PR was closed as superseded by #19817. The regression on #19817 now
publishes explicitly to initial partitions 0, 1, and 2 and asserts each
per-partition end offset. It also explicitly covers the newly added partitions
after expansion. See commits cb6d8b2e66 and 3b49ef3f3f.
--
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]
