fix(test): wait for Kafka partitions before publishing

[FrankChen021]

Description

This fixes a race in the embedded Kafka test resource that was exposed by the failed KafkaIndexFaultToleranceTest run in #19754. The failure is unrelated to that PR's dependency changes.

After creating a topic or increasing its partition count, the Kafka admin operation can complete before the partition leaders are ready to serve requests. The test helper immediately created an idempotent producer and published records, which could receive NOT_LEADER_OR_FOLLOWER followed by repeated OUT_OF_ORDER_SEQUENCE_NUMBER responses. Because the helper ignored the futures returned by KafkaProducer.send(), it could return after silently dropping records; the ingestion test then timed out waiting for rows that never reached Kafka.

This PR:

• waits for a successful latest-offset lookup through every partition leader after creating a topic or increasing its partition count;
• waits for every non-transactional producer send result, so delivery failures are surfaced;
• adds a regression test that publishes immediately after both topic creation and partition expansion and verifies that all records arrive.

There is no end-user behavior change.

Key changed classes

• KafkaResource
• KafkaResourceTest

This PR has:

• been self-reviewed.
• added comments explaining the "why" and the intent of the code wherever it would not be obvious.
• added or updated tests to cover the changed behavior.
• run the exact embedded test that originally failed.

Tests

mvn -ntp test -pl extensions-core/kafka-indexing-service -am \
  -Dtest="org.apache.druid.indexing.kafka.simulate.KafkaResourceTest" \
  -Dsurefire.failIfNoSpecifiedTests=false -Pskip-static-checks \
  -Dweb.console.skip=true -T1C

mvn -ntp test -pl embedded-tests -am \
  -Dtest="org.apache.druid.testing.embedded.indexing.KafkaIndexFaultToleranceTest#test_supervisorRecovers_afterChangeInTopicPartitions" \
  -Dsurefire.failIfNoSpecifiedTests=false -Pskip-static-checks \
  -Dweb.console.skip=true -T1C

mvn -ntp test -pl embedded-tests -am \
  -Dtest="org.apache.druid.testing.embedded.indexing.autoscaler.CostBasedAutoScalerIntegrationTest#test_autoScaler_scalesUpAndDown_withSlowPublish" \
  -Dsurefire.failIfNoSpecifiedTests=false -Pskip-static-checks \
  -Dweb.console.skip=true -T1C

All pass locally. The fault-tolerance test runs both transactional and non-transactional parameterizations.

[Copilot]

Pull request overview

This PR hardens the embedded Kafka test resource used by Druid’s ingestion/embedded tests by eliminating races around partition expansion and by surfacing producer delivery failures that were previously silently ignored.

Changes:

• After increasing a topic’s partition count, KafkaResource#increasePartitionsInTopic now blocks until each partition can successfully serve a latest-offset lookup via its leader.
• KafkaResource#produceRecordsWithoutTransaction now waits on KafkaProducer.send() futures so publish failures are not silently dropped.
• KafkaResourceTest adds a regression scenario that increases partitions and immediately publishes records, validating end offsets.

Reviewed changes

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

File	Description
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/KafkaResource.java	Adds leader-readiness verification after createPartitions() and waits for non-transactional send results to surface delivery failures.
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/KafkaResourceTest.java	Adds a regression check for publishing immediately after partition expansion.

---

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

[Copilot]

Pull request overview

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

Comments suppressed due to low confidence (2)

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/KafkaResource.java:330

• The retry predicate relies on operator precedence (|| vs &&) to recurse into the cause chain. Adding parentheses makes the intended logic unambiguous and avoids misreads during maintenance.
  extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/KafkaResource.java:318
• The local variable name partitionOffsets is misleading here: the map holds OffsetSpec requests, not actual offsets. Renaming it to reflect its contents makes the readiness check easier to understand/maintain.

    admin.listOffsets(partitionOffsets).all().get();
  }

  private Map<String, Object> commonClientProperties()

[Copilot]

Pull request overview

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

[FrankChen021]

Reviewed 2 of 2 changed files plus surrounding call sites. No actionable findings; AdminClient retries partition-readiness lookups and producer futures are awaited.

---

This is an automated review by Codex GPT-5.6-Sol