[kafka] Implement Metadata API over Fluss table metadata - #4261
Draft
gyang94 wants to merge 5 commits into
Draft
Conversation
Introduce API registration, request context, version validation, and asynchronous error mapping. Fix request buffer ownership and response serialization cleanup while preserving the existing ApiVersions entry point. Validated with mvn -o -pl fluss-rpc,fluss-kafka verify. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 555/555 AI-Contributed/UT: 525/525
Route requests through the dispatcher and advertise only implemented APIs. Return version-aware errors for unsupported APIs and invalid requests. Validated with mvn -o -pl fluss-kafka verify (23 unit tests and 1 IT). Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 313/313 AI-Contributed/UT: 137/137
Add Metadata v0-v11 handling over pre-created Fluss tables, with listener-aware gateway access, table/topic mapping, partition metadata, and error handling. Do not implement CreateTopics, DeleteTopics, or automatic table creation. Honor authoritative ISR values when bucket epochs are present. For legacy metadata without an epoch, conservatively report only an available leader. Verified fluss-kafka: 37 unit tests and 1 integration test, with Spotless and Checkstyle passing. Rebuilt the matching RPC dependency from the same base. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 909/909 AI-Contributed/UT: 540/540
This was referenced Sep 9, 2026
Extract topic identity and the raw/string table mapping contract before Metadata. Validate table kinds, field projections and metadata columns independently of request handling and record decoding. Validation: Java 11, mvn -o -pl fluss-kafka clean verify (47 unit tests and 2 integration tests); Checkstyle, Spotless and RAT passed. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 769/769 AI-Contributed/UT: 480/480
Integrate the DDL mapping prerequisite and use its resolver for Metadata discovery. Omit unsupported tables from all-topic queries and return per-topic mapping errors for named queries. Preserve metadata for compatible tables in mixed requests. Validation: Java 11, mvn -o -pl fluss-kafka clean verify (62 unit tests and 2 integration tests); Checkstyle, Spotless and RAT passed. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 742/742 AI-Contributed/UT: 571/571
This was referenced Sep 10, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
Closes #4266. Part of #4185.
Expose pre-created Fluss tables through Kafka Metadata using the shared DDL mapping contract. Compatible tables return routing metadata; incompatible mappings are omitted from all-topic discovery and produce a per-topic error in named queries, while other valid topics remain discoverable.
This is PR04 of the foundation sequence, depending on PR03 DDL table mapping (#4277 / #4275), which builds on PR02 ApiVersions (#4260 / #4265) and PR01 (#4259 / #4264).
Brief change log
table_jsonmetadata field.INVALID_TOPIC_EXCEPTIONfor incompatible mappings andUNKNOWN_TOPIC_OR_PARTITIONfor missing tables, without suppressing valid topics in the same request.Relative to the DDL prerequisite: 9 files, +1477 / -4 lines (1481 changed lines including tests). Shared DDL mapping implementation and tests belong to #4277 and are excluded from this incremental scope. The Metadata protocol, gateway adaptation and version/error regression coverage are kept together as one review unit.
Tests
Validated on Java 11 at commit
050b26088ba0e9fbd7b5db7851da1688d6715b73:mvn -o -pl fluss-kafka clean verify: 62 unit tests and 2 integration tests passed.git diff --checkpassed. Matching dependencies were built/installed withmvn -o -pl fluss-kafka -am install -DskipTests.The full repository test suite and a Flink SQL runtime were not run.
API and Format
Adds Kafka Metadata v0-v11 for tables that satisfy the current DDL mapping contract. Existing Fluss RPC and storage formats are reused. Kafka CreateTopics/DeleteTopics, automatic table creation, Metadata v12+, Produce, and JSON record conversion remain outside this PR.
Documentation
The supported versions, DDL admission rules and per-topic errors are described in this PR, source Javadoc and regression tests. Produce delivery documentation follows with later capability work.
Stack and review
This PR remains a draft until its prerequisites land. It targets Apache
main, so GitHub's Files changed view includes the foundation prerequisites. Review Metadata alone using the incremental comparison against DDL mapping. The DDL prerequisite is integrated without rewriting the previously published Metadata history. Update the baseline after prerequisites merge and validate again before marking ready.Generative AI disclosure: Codex assisted with implementation, testing, and commit organization. Human review is required before merge.