[kafka] Implement Metadata API over Fluss table metadata

[gyang94]

Purpose

Closes #4266. Part of #4185.

Expose pre-created, Kafka-enabled Fluss tables through Metadata v0-v11. Clients can address a table by its canonical database.table name or its configured default-database alias, cache both names together, and detect table recreation. Missing databases and hidden tables do not suppress brokers or other valid topics.

This is PR04, based on PR03 DDL table mapping (#4277 / #4275), following PR02 ApiVersions (#4260 / #4265) and PR01 (#4259 / #4264).

Brief change log

• Implement and advertise Metadata v0-v11 through a narrow backend contract and a local TabletServer gateway adapter. Reject Metadata v12 before invoking the gateway and ID-only lookup in v10/v11.
• Resolve canonical names and aliases using the configured default database (fluss by default). Named responses retain requested names; all-topic discovery lists canonical names only. Missing aliases never fall back to another database.
• Reuse PR03's public schema contract and persisted table.kafka.enabled opt-in. Disabled, missing and authorization-filtered tables return UNKNOWN_TOPIC_OR_PARTITION without exposing an ID. Enabled tables with invalid contracts return INVALID_TOPIC_EXCEPTION. Persisted enablement changes remain visible without a stale schema cache.
• Give aliases stable wire UUIDs in a separate namespace derived from the same native Fluss Table ID. This supports Kafka client caches containing both names and recreation with a lower leader epoch. Both names retain the same table, buckets, leaders and native authorization identity.
• Preserve brokers, aliases and valid topics across missing-database or deletion races, including synchronous and asynchronous gateway failures. Propagate unrelated failures.
• Map partitions, leaders, replicas, authoritative ISR and offline replicas using the Kafka listener. Use a conservative available-leader ISR fallback only when bucket epoch is absent.
• Pass one RpcGatewayService through the plugin and request handler. The backend checks that it implements TabletServerGateway and casts it internally, ensuring both typed references refer to the same service.
• Consolidate overlapping Metadata tests while preserving version coverage, alias/canonical identity, real Kafka cache updates, authorization filtering, missing-database retries and real listener behavior.
• Keep the snapshot-lease restart regression fix: wait for server metadata to converge after each coordinator restart while leaving the client cache stale to exercise refresh and retry.

The latest PR03 creates a configured custom Kafka default database at coordinator startup when Kafka is enabled. The native fluss database retains its cluster-initialization behavior. Metadata requests themselves create neither databases nor tables and do not recreate a deleted default database; the integration test now reflects this distinction.

Increment relative to PR03: 14 files changed, 2465 insertions(+), 19 deletions(-). DDL validation and startup database creation remain in PR03.

Tests

Validated the tree committed as ecbb368832efd3d5f542880c190e419f3ad5604b with Java 11:

mvn -o -pl fluss-kafka,fluss-client -am clean verify \
  '-Dtest=Kafka*Test,Kafka*ITCase,ApiVersionsHandlerTest,RetryableGatewayClientProxyTest,TableChangeWatcherTest,FlussAdminITCase#testKvSnapshotLease*' \
  -Dsurefire.failIfNoSpecifiedTests=false

• All 158 distinct selected cases passed with zero failures, errors or skips: 112 Kafka, 28 public schema-contract, 6 RPC retry, 5 default-database, 5 table-change watcher and 2 snapshot-lease cases. Both Maven test phases passed; counts do not double-count them.
• Kafka coverage includes 31 Metadata handler cases, 22 mapper cases, 5 real Metadata integration cases and 12 table-mapping integration cases.
• Both snapshot-lease cases passed, including the previously failing testKvSnapshotLeaseAfterCoordinatorServerRestart.
• All Fluss SNAPSHOT test dependencies came from this worktree's clean reactor build. Spotless, Checkstyle, RAT and git diff --check passed.

Real Kafka Metadata cache tests cover v10/v11 full and partial updates with canonical names and aliases together, plus delete/recreate with leader epoch dropping from 5 to 0. Listener tests exercise Metadata v0-v11, default/custom aliases, startup-created and deleted default databases, persisted enablement changes, invalid enabled contracts and canonical-only discovery. The optional three-field _kafka_metadata contract is covered through named and discovery paths; the legacy fourth offset field is rejected.

The full repository and full Admin suites, Flink SQL runtime and actual Producer writes were not run. The earlier Java 11/core CI failure in RebalanceITCase.testRebalanceForLogTable is outside the Metadata increment and is not fixed or covered by this validation.

API and Format

Uses existing Fluss RPC and storage formats. The optional _kafka_metadata column follows PR03: timestamp TIMESTAMP_LTZ(3), timestamp_type INT and partition INT. Both Kafka UUID namespaces decode to the same native Fluss Table ID.

CreateTopics/DeleteTopics, automatic table creation, Metadata v12+, Produce and record conversion are outside this PR. No PR07 implementation or tests are included. Authentication/session propagation and service-disabled protocol semantics remain in later PR16/19 and PR20 respectively.

Documentation and review

Supported versions, naming, discovery, identity and errors are documented here, in Javadoc and in regression tests.

The exact PR03 baseline is 7bcda8b4a17282b9e956137ea31ae281cbf5795b. Review Metadata alone using the incremental comparison. This PR targets Apache main, so Files changed also contains unmerged prerequisites. Revalidate the stack after prerequisites merge.

Generative AI disclosure: Codex assisted with implementation, testing and commit organization. Human review is required before merge.