Copilot commented on code in PR #19824:
URL: https://github.com/apache/druid/pull/19824#discussion_r3686687854
##########
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:
`createTopics(...).all().get()` can return before leaders are ready, and
`listOffsets(...).all().get()` can also fail transiently (e.g., retriable
metadata/leader errors). Since the intent here is to *wait* for readiness, it’s
safer to retry `listOffsets` for a bounded number of attempts rather than
failing immediately and keeping the test flaky.
##########
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 regression publishes 1,000 records without ensuring they hit all three
partitions. With a null key, Kafka’s default partitioner can legally send all
records to a single partition, so this test may pass without exercising every
partition leader (and it doesn’t match the PR description about reaching the
three partitions). Consider producing records explicitly to partitions 0/1/2
and asserting per-partition end offsets.
--
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]
