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]

[P2] Wait for the catch-up worker before disposing its cache

When a finite connectTimeout expires while this new catch-up loop is processing a backlog, start() calls future.cancel(true) and immediately cacheHandler.close(). Cancellation only interrupts the worker; the record loop has no interrupt check, so it can continue put/remove calls after close. With the off-heap cache manager, close deletes the underlying MapDB map, racing those writes against deletion. Coordinate worker termination, or move cache disposal into the worker's completion path, before closing the cache.

[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

[FrankChen021]

[P1] Do not wait forever after connectTimeout expires

After start() reaches its configured timeout, this call waits indefinitely for executor termination. future.cancel(true) only requests interruption and cannot guarantee that consumer code exits; the updated test demonstrates the regression by configuring a 1 ms timeout yet requiring start() to remain blocked until an uninterruptible poll is manually released. LookupReferencesManager.startLookups() waits for every lookup result, so one stuck Kafka consumer can now block lookup-manager and process startup indefinitely instead of returning false after connectTimeout. Preserve cache safety without synchronously waiting forever, for example by transferring cache disposal to worker completion while allowing start() to return after bounded cancellation.

[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