FrankChen021 commented on code in PR #19824:
URL: https://github.com/apache/druid/pull/19824#discussion_r3686804682
##########
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:
No additional outer retry is needed in the surviving implementation. Kafka
4.3 ListOffsetsHandler already remaps and retries NOT_LEADER_OR_FOLLOWER and
LEADER_NOT_AVAILABLE responses and retries other RetriableException results
within the Admin API timeout. PR #19817 keeps the single listOffsets readiness
barrier and avoids duplicating that retry layer.
##########
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:
The test-strengthening point is incorporated in PR #19817 commit 3fc8dfab92:
records are published explicitly to every initial and newly added partition and
each partition end offset is asserted. PR #19824 is being retired as the
narrower duplicate.
--
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]
