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.