fix: Wait for Kafka lookup startup catch-up

[sakshichitnis27]

Fixes #19575

Description

KafkaLookupExtractorFactory previously reported a lookup as started immediately after the consumer's first poll(), even when the consumer had not finished reading the existing topic data. This allowed newly started brokers to serve queries using a partially populated lookup.

This change:

• Captures the end offsets after the consumer receives its initial partition assignment.
• Processes polled records before checking catch-up progress.
• Releases the startup latch only after every assigned partition reaches its captured end offset.
• Refreshes the captured offsets if the partition assignment changes during startup.
• Stops checking offsets after startup completes.

The existing connectTimeout continues to bound how long start() waits. A value of 0 retains the existing do-not-wait behavior.

A regression test verifies that start() remains blocked while the consumer is behind and succeeds after all initial records have been applied to the lookup.

Testing

mvn test -pl extensions-core/kafka-extraction-namespace -am \
  -Dtest=org.apache.druid.query.lookup.KafkaLookupExtractorFactoryTest \
  -Dsurefire.failIfNoSpecifiedTests=false \
  -Dweb.console.skip=true \
  -T1C

[FrankChen021]

Severity	Findings
P0	0
P1	0
P2	1
P3	0
Total	1

Reviewed 2 of 2 changed files. Found one startup-timeout cleanup race where the Kafka worker may keep writing after its cache is disposed.

---

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

[FrankChen021]

Severity	Findings
P0	0
P1	1
P2	0
P3	0
Total	1

Severity	Findings
P0	0
P1	1
P2	0
P3	0
Total	1

Reviewed 2 of 2 changed files. The prior cache-disposal race is resolved, but the replacement cleanup path can block process startup indefinitely after the configured timeout.

---

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

[sakshichitnis27]

@FrankChen021
Thanks for catching this. I pushed a follow-up fix in d7d6f3b.
The timeout path no longer waits indefinitely for executor termination. Cache disposal is queued behind the Kafka worker on the same single-thread executor, so start() can return false after the timeout while the cache remains available until the worker has actually stopped using it.
I also updated the regression test to verify that:

• start() returns while an uninterruptible poll is still blocked;
• the cache is not disposed prematurely; and
• cleanup completes after the worker exits.
  Locally, all 18 KafkaLookupExtractorFactoryTest tests pass, including five consecutive full-class runs, along with Checkstyle, PMD, Maven Enforcer, forbidden-API checks, and compilation.

[FrankChen021]

I have reviewed the updated head for correctness, edge cases, concurrency, and integration risks; no issues found.

Reviewed 2 of 2 changed files. The prior unbounded-startup-wait finding is resolved while cache disposal remains ordered after the Kafka worker exits.

---

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

[FrankChen021]

@sakshichitnis27 could you resolve the conflict?

[sakshichitnis27]

@FrankChen021 Thanks! I’ve resolved the conflict by merging the latest master into the PR branch. I also reran the targeted Kafka lookup tests locally; all 18 tests passed, along with Checkstyle, PMD, Maven Enforcer, compilation, and forbidden-API checks. The PR is now mergeable. GitHub Actions are currently waiting for maintainer approval to run. Please take another look when you have a chance.

[FrankChen021]

🟢 Approval recommended

No actionable issues found in the current head. The startup barrier captures the assigned partitions' end offsets, applies polled records before releasing startup, and refreshes the target set when the assignment changes. The timeout cleanup path queues cache disposal behind the single Kafka worker, so the prior cache-write race and unbounded startup wait are resolved.

Reviewed exactly 2 of 2 changed files:

• extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java
• extensions-core/kafka-extraction-namespace/src/test/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactoryTest.java

Static review only; no builds, tests, installs, or formatters were run.

---

This is an automated review by Codex GPT-5.6-Luna(max)

[FrankChen021]

🟢 Approval recommended

No actionable issues found in the current head. The incremental update only replaces deprecated MockConsumer constructors in the regression tests. I rechecked the full current implementation and surrounding cache/consumer lifecycle: startup applies polled records before releasing the captured-offset barrier, assignment changes refresh the tracked partitions, and shutdown keeps cache disposal behind the single Kafka worker so the timeout path does not dispose a cache that the worker can still mutate.

Reviewed 2 of 2 changed files:

• extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java
• extensions-core/kafka-extraction-namespace/src/test/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactoryTest.java

Validation: git diff --check 15051a5c09b04eda967dee9dbc552a6dc658d794..HEAD passed. Static review only; no builds, tests, installs, or formatters were run.

---

This is an automated review by Codex GPT-5.6-Luna(max)

[FrankChen021]

@sakshichitnis27 Still some warnings from the github-advanced-security

[sakshichitnis27]

@FrankChen021 Thanks for pointing this out. I found three remaining deprecated MockConsumer(OffsetResetStrategy) usages and replaced all of them with the supported string constructor in f116b51. I also removed the unused deprecated enum import.
The relevant build and all 18 tests pass locally, including five repeated runs. Fresh GitHub Actions and CodeQL runs are now waiting for maintainer approval. Please take another look when they complete.

[sakshichitnis27]

Hi @FrankChen021, I investigated the failed CI job. The only failure was the unrelated KillSupervisorsCustomDutyTest, where supervisor history transiently returned 404.
I reran the exact test on JDK 25: 1 test passed, 0 failures. Its complete 39-module reactor also passed with BUILD SUCCESS. I separately reran KafkaLookupExtractorFactoryTest: all 18 tests passed with BUILD SUCCESS.
Could you please rerun the failed job? I have avoided changing the unrelated embedded test so the PR remains minimal and focused.