fix(test): wait for Kafka topic leaders after creation

[FrankChen021]

Description

This fixes the embedded Kafka startup race exposed by
KafkaIndexFaultToleranceTest.test_supervisorRecords_afterHistoricalRestart in #19808.

Admin.createTopics(...).all().get() confirms that the topic metadata operation completed, but it does not guarantee
that every partition leader is already ready to serve requests. StreamIndexFaultToleranceTest creates a topic and
immediately starts its supervisor and publishes records. If the initial partition leaders are still becoming
available, the first batch can fail to reach Kafka and the test waits for an ingest/rows/published event that never
arrives.

This PR:

• verifies a latest-offset lookup through every initial partition leader before createTopicWithPartitions returns;
• adds regression coverage that publishes 1,000 non-transactional records immediately after topic creation and
  verifies all records reached the three partitions.

This complements #19817: that PR covers readiness after increasing an existing topic's partition count and surfaces
asynchronous producer failures; this PR covers readiness immediately after initial topic creation.

There is no end-user behavior change. The affected helper is used only by embedded tests.

Tests

• KafkaResourceTest passed with the immediate non-transactional publish regression.
• KafkaIndexFaultToleranceTest#test_supervisorRecords_afterHistoricalRestart passed twice in focused embedded-test
  reactor runs.
• git diff --check passed.

Created by GPT-5.6-Sol.

[Copilot]

Pull request overview

Improves stability of embedded Kafka-based tests by reducing a post-topic-creation race where Kafka admin topic creation can complete before partition leaders are able to serve requests, which can cause immediate publish/supervisor startup to fail intermittently.

Changes:

• Adds a post-createTopics readiness check in KafkaResource#createTopicWithPartitions via a latest-offset lookup for each partition.
• Extends KafkaResourceTest with a regression scenario that publishes immediately after topic creation and validates end offsets.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/KafkaResource.java	Adds a leader-readiness validation step after topic creation using Admin#listOffsets.
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/KafkaResourceTest.java	Adds a regression check that publishing immediately after topic creation results in the expected end offsets.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[FrankChen021]

Closing this as a duplicate of #19817. Since this PR was opened, #19817 has been expanded to wait for Kafka partition leaders after both initial topic creation and partition expansion, and it also waits for non-transactional producer send results so delivery failures are surfaced. Its validation includes the exact CostBasedAutoScalerIntegrationTest#test_autoScaler_scalesUpAndDown_withSlowPublish flake. The current #19817 therefore contains the functional fix proposed here plus the broader partition-expansion and producer-error handling.

[FrankChen021]

Superseded by #19817, which now covers initial topic creation, partition expansion, non-transactional send completion, and explicit per-partition regression coverage. The useful Copilot test feedback from this PR was incorporated there and validated with KafkaResourceTest plus both partition-change fault-tolerance parameterizations.