[kafka] Define DDL table mapping for Kafka compatibility by gyang94 · Pull Request #4277 · apache/fluss

gyang94 · 2026-09-10T07:55:35Z

Purpose

Closes #4275. Part of #4185.

Define the shared DDL table mapping contract used by Metadata and Produce. A Kafka topic named database.table resolves directly to that Fluss table, so tables with the same name in different databases remain distinct. The physical schema and kafka.* custom properties define how record fields map to table columns.

This is PR03 in the foundation sequence, based on PR02 (#4260, commit 5742b36627fe156a8cf499f0aa7b4b8c4163e9a7); Metadata follows as PR04 (#4261).

Brief change log

Map fully qualified topic names in both directions and derive stable topic IDs from Fluss table IDs. Validate both Fluss component names and Kafka's full topic-name length limit.
Validate non-partitioned Arrow log tables without primary keys and resolve raw/string key and value formats.
Resolve key/value field projections and optional timestamp/headers columns; reject invalid types, missing fields, overlapping projections, and unsupported mapping options. Build projected field names without repeatedly allocating the full schema's field-name list.
Cover naming boundaries, schema validation, and native Fluss create-table → metadata-readback behavior, including same-named tables in different databases. Reuse the existing curator-test version in test scope for the cluster fixture.

Relative to PR02: 10 files, +1374 / -0 lines (791 production/build and 583 test lines). The change keeps the mapping contract and its validation together. Record decoding, Arrow encoding, Metadata handlers, Produce, JSON, and authentication follow in later PRs.

Tests

Validated on Java 11 with the source tree committed as 4361fa11b227f6d5573757b21499f692a289cc12:

67 Kafka unit tests passed using the same-checkout Maven reactor.
KafkaRequestITCase and KafkaTableMappingITCase passed: 2 integration tests, including cross-database mapping and native metadata readback.
Checkstyle, Spotless, RAT, and git diff --check passed.

The full repository test suite and a Flink SQL runtime were not run. The later delivery stage will cover the full DDL → Metadata → Kafka producer → Fluss readback workflow.

API and Format

Uses the existing table schema and custom-property storage. kafka.value.format is required; key mapping is optional. Each mapped raw or string component requires exactly one BYTES or STRING column respectively. Configured key fields require kafka.value.fields-include=EXCEPT_KEY; timestamp and headers use explicitly named physical columns and are excluded from the value projection. Column names are user-defined.

Topic names must use database.table; bare table names do not fall back to a configured database. Reverse mapping returns the same fully qualified name.

No Kafka API is newly advertised by this PR. Table creation continues through existing Fluss DDL/Admin or the Flink catalog; the resolver validates the stored mapping when called by subsequent API layers. Kafka CreateTopics/DeleteTopics and automatic creation are out of scope.

Documentation

The contract is described in this PR, source Javadoc, and executable tests. User-facing delivery examples follow with the later Produce capability.

Stack and review

[kafka] Add request dispatch and transport framework #4259 — PR01: request and transport framework.
[kafka] Serve ApiVersions from registered capabilities #4260 — PR02: ApiVersions.
This PR — PR03: DDL table mapping ([kafka] Define DDL table mapping for Kafka compatibility #4275).
[kafka] Implement Metadata API over Fluss table metadata #4261 — PR04: Metadata ([kafka] Implement Metadata API for pre-created Fluss tables #4266).

This PR remains a draft while its prerequisites are unmerged. It targets Apache main, so the Files changed view includes the foundation prerequisites. Review DDL mapping alone using the incremental comparison against PR02.

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

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

Resolve database.table topic names directly to Fluss table paths and return qualified names for metadata. Validate name boundaries and cover same-named tables in different databases through native metadata round trips. Read projected field names directly from DataField to avoid rebuilding the complete field-name list for every projected column. Validated on Java 11: 67 Kafka unit tests and 2 integration tests passed, along with Checkstyle, Spotless, RAT, and git diff --check. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 52/52 AI-Contributed/UT: 167/167

Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 16/16 AI-Contributed/UT: 33/33

Wait for the newly created log table to become ready before acquiring or releasing snapshot leases. This prevents the coordinator's asynchronous table-ID registration from masking the expected NonPrimaryKeyTableException. Validated on Java 11 with the same-worktree reactor: FlussAdminITCase#testKvSnapshotLeaseForLogTable passed, along with Checkstyle, Spotless, RAT, and git diff --check. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 0/0 AI-Contributed/UT: 1/1

Register table.kafka options and validate the declared fixed-column mapping through a shared schema contract. Support binary, string and flat/nested JSON values with optional key, headers and rescue columns. Validate CREATE and ALTER before persistence, and fence schema publication with the table version and coordinator epoch so concurrent schema and property changes cannot bypass the contract. Resolve bare topic aliases through kafka.default.database and stop automatically creating that database. Keep real offset persistence deferred and reject Kafka enablement for tables that declare offset metadata until that capability is available. Validation: 146 targeted tests passed on Java 11; Checkstyle, Spotless, RAT and git diff checks passed. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 1131/1131 AI-Contributed/UT: 1451/1451

Default bare topic names to the native fluss database while preserving explicit database overrides, canonical names and no automatic Kafka database creation. Validate JSON business types recursively before CREATE, ALTER and enablement. Reject non-STRING map keys at every nesting level and schemas deeper than the supported limit. Expose the validation entry point and depth constant for subsequent decoders. Count business fields equally in flat and nested layouts, without charging a level for the physical _kafka_value ROW container. Keep disabled tables outside Kafka validation and preserve user schemas and explicit properties. Cover complex scalar/container types, depth boundaries, failed DDL state preservation and successful enablement with real RPC tests. Validation: Java 11 clean reactor verify passed 161 targeted cases, including snapshot lease and concurrent ALTER regressions. Checkstyle, Spotless, RAT and git diff checks passed. The five prior PR03 commits were rebased onto PR02 97d20b1 and checked with range-diff. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 92/92 AI-Contributed/UT: 292/292

Allow optional _kafka_metadata with timestamp, timestamp_type and partition, without a physical offset field. Cover optional fixed-column combinations, metadata validation and CREATE/ALTER behavior. Ignore unchanged table registrations when schema ALTER advances the ZK version. Preserve atomic concurrency fencing and notifications for real property and bucket changes, with deterministic regression coverage. Validation: Java 11 reactor verify passed 111 targeted test cases; Checkstyle, Spotless and git diff --check passed. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 57/57 AI-Contributed/UT: 265/265

Create the configured Kafka default database when missing and preserve existing metadata. Fence schema and property ALTER with schema-parent version checks without rewriting table registrations, restoring the original table watcher. Remove redundant configuration, Flink conversion and mapping test coverage while retaining DDL validation, optional system columns and concurrent ALTER checks. Validation: targeted clean verify passed 173 tests with Checkstyle and Spotless. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 120/120 AI-Contributed/UT: 683/683

Reuse native initialization for the fluss database and create a custom Kafka default database only when it is missing. Remove PR03-only schema version and Coordinator epoch checks, preserving PR02 metadata writes and existing protections. Drop the corresponding concurrency tests while retaining Kafka DDL mapping validation. Validation: 121 focused tests passed; Maven verify checks passed. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 132/132 AI-Contributed/UT: 224/224

State that Kafka access requires an existing Fluss table and summarize qualified topic mapping and the default database fallback. Validation: git diff --check. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 13/13 AI-Contributed/UT: 0/0

Describe topic mapping before the requirement to create its Fluss table. Validation: git diff --check. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 2/2 AI-Contributed/UT: 0/0

wuchong · 2026-10-17T07:57:21Z

+                || new TableConfig(Configuration.fromMap(table.getProperties())).getLogFormat()
+                        != LogFormat.ARROW) {


Why do we have this limitation?

wuchong

Thanks @gyang94 , I have finished my review.

Use typed conversion options with read-time defaults and keep all four system columns optional. Validate nullable fields recursively while preserving Fluss's non-null MAP key requirement and allow nested business fields to use system names. Ensure the configured Kafka default database exists, consolidate table validation entry points, and document the mapping contract in a dedicated Kafka compatibility guide with regression coverage. Validation: 102 targeted tests passed; after the enum rename, 52 related tests and the affected Maven reactor verification passed. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 648/648 AI-Contributed/UT: 492/492

Use Kafka-prefixed value conversion enums and keep path-aware table descriptor validation under its established method name. Simplify the user guide and show topic-to-table examples in a compact mapping table. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 315/315 AI-Contributed/UT: 22/22

Reorder the conversion enum imports and wrap the long test builder call to match the repository's Spotless format. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 2/2 AI-Contributed/UT: 4/4

Co-Authored-By: Codex <noreply@openai.com> AI-Contributed/Feature: 48/48 AI-Contributed/UT: 0/0

wuchong · 2026-10-18T16:29:14Z

I removed some AI slop. Please review AI-generated code carefully yourself before submitting it.

gyang94 force-pushed the codex/kafka-ddl-table-mapping branch from 6b9626d to 4361fa1 Compare September 17, 2026 03:40

gyang94 force-pushed the codex/kafka-ddl-table-mapping branch from 4361fa1 to c508de6 Compare October 10, 2026 13:19

gyang94 marked this pull request as ready for review October 12, 2026 01:39

gyang94 force-pushed the codex/kafka-ddl-table-mapping branch 3 times, most recently from d661491 to e7673be Compare October 14, 2026 06:11

gyang94 and others added 9 commits October 17, 2026 11:13

[kafka] Define Kafka timestamp metadata as TIMESTAMP(3)

1082ed6

Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 16/16 AI-Contributed/UT: 33/33

gyang94 force-pushed the codex/kafka-ddl-table-mapping branch from 7bcda8b to 7badab0 Compare October 17, 2026 03:16

gyang94 and others added 2 commits October 17, 2026 15:28

[docs] Reorder Kafka topic mapping explanation

92522a9

Describe topic mapping before the requirement to create its Fluss table. Validation: git diff --check. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 2/2 AI-Contributed/UT: 0/0

wuchong reviewed Oct 17, 2026

View reviewed changes

Comment thread fluss-common/src/main/java/org/apache/fluss/metadata/KafkaTableSchema.java

Comment thread fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java Outdated

gyang94 and others added 4 commits October 18, 2026 18:11

[kafka] Apply Spotless formatting fixes

b98ca1c

Reorder the conversion enum imports and wrap the long test builder call to match the repository's Spotless format. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 2/2 AI-Contributed/UT: 4/4

remove AI slop

3af2b1e

Co-Authored-By: Codex <noreply@openai.com> AI-Contributed/Feature: 48/48 AI-Contributed/UT: 0/0

Navigation Menu

Repository navigation

[kafka] Define DDL table mapping for Kafka compatibility - #4277

[kafka] Define DDL table mapping for Kafka compatibility#4277
gyang94 wants to merge 15 commits into
apache:mainfrom
gyang94:codex/kafka-ddl-table-mapping

gyang94 commented Sep 10, 2026 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

wuchong Oct 17, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

wuchong left a comment

Uh oh!

Uh oh!

Uh oh!

wuchong commented Oct 18, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		\|\| new TableConfig(Configuration.fromMap(table.getProperties())).getLogFormat()
		!= LogFormat.ARROW) {

Conversation

gyang94 commented Sep 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Brief change log

Tests

API and Format

Documentation

Stack and review

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

wuchong Oct 17, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

wuchong commented Oct 18, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

gyang94 commented Sep 10, 2026 •

edited

Loading