Skip to content

[kafka] Transcode mapped raw and string records to Arrow - #4285

Open
gyang94 wants to merge 21 commits into
apache:mainfrom
gyang94:codex/kafka-raw-string-transcode
Open

[kafka] Transcode mapped raw and string records to Arrow#4285
gyang94 wants to merge 21 commits into
apache:mainfrom
gyang94:codex/kafka-raw-string-transcode

Conversation

@gyang94

@gyang94 gyang94 commented Sep 10, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Purpose

Closes #4282. Part of #4185.

Use the DDL mapping contract to decode raw/string keys and values, assemble physical rows with optional timestamps and ordered headers, and encode owned native Arrow log bytes.

This is PR06, based on the latest #4284 (PR05) at ff27c55c709bf1386b6f45d549d7de2b8525b064.

Brief change log

  • Raw bytes and strict UTF-8 strings map to the configured physical columns.
  • Nullable values, empty bytes, NOT NULL violations and duplicate/nullable headers are handled correctly.
  • The native batch uses the table schema ID and remains readable after Arrow resources close.
  • Decode, assemble and append records synchronously one at a time, avoiding an intermediate list of all physical rows.
  • Borrow command-owned key, value and header payloads during conversion, preserving defensive-copy constructors and accessors.
  • Arrow writers and allocators close on encoding or source failures, including failures after rows have already been appended.

Incremental scope relative to PR05: 8 files, +923 / -0 (373 implementation and 550 test lines), all in fluss-kafka. The line count includes Javadoc and license headers. No Fluss core changes are required for these optimizations.

Tests

Java 11 at 04a9093b331cfb1835493627ef0ab0adb969b8c2:

  • mvn -o -pl fluss-kafka -am clean verify '-Dtest=*Kafka*Test,Kafka*ITCase' -Dsurefire.failIfNoSpecifiedTests=false: 130 unit tests and 4 integration tests passed.
  • Checkstyle, Spotless, RAT and git diff --check passed.
  • Regression tests cover defensive-copy ownership, null/empty payloads, raw/string batches beyond initial Arrow capacity, row and payload reuse, and cleanup after partial conversion failures.
  • An additional 25 local boundary scenarios passed, covering compression modes, multiple output pages, timestamp extremes, malformed UTF-8, failure paths and concurrent calls.
  • Production Produce registration and append integration remain in PR07.

The full repository test suite and Flink SQL runtime were not run.

API and Format

Adds raw/string-to-Arrow conversion for the existing DDL contract, with no wire or storage format change. JSON decoding and append integration remain separate.

Stack and review

Draft while prerequisites are unmerged. The PR targets Apache main, so the full Files changed view includes prerequisites. Use this incremental comparison to review this stage alone.

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

This was referenced Sep 10, 2026
Open
Open
@gyang94
gyang94 force-pushed the codex/kafka-raw-string-transcode branch from 5031362 to 04a9093 Compare September 17, 2026 09:20
@gyang94
gyang94 force-pushed the codex/kafka-raw-string-transcode branch from 04a9093 to 41981bc Compare October 14, 2026 16:48
@gyang94
gyang94 marked this pull request as ready for review October 15, 2026 04:15
@gyang94
gyang94 force-pushed the codex/kafka-raw-string-transcode branch from 41981bc to ef27192 Compare October 15, 2026 11:33
gyang94 and others added 21 commits October 19, 2026 10:22
@gyang94 @codex
[kafka] Use the DDL table contract in Metadata
744ac0c 
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
@gyang94 @codex
[kafka] Discover qualified topics and isolate missing databases
c0c3816 
Route Metadata queries through the database.table mapping from PR03 and
enumerate compatible topics across databases. Preserve brokers and valid
topics when a requested database is absent or disappears during discovery;
propagate other gateway failures.

Cover qualified names, same-named tables across databases, synchronous and
asynchronous lookup failures, and Metadata v0-v11 over a real listener.

Validated on Java 11: 90 unit tests and 4 integration tests passed, along
with Checkstyle, Spotless, RAT, and git diff --check. Five additional
interleaved asynchronous review probes passed.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 129/129
AI-Contributed/UT: 514/514
@gyang94 @codex
[kafka] Preserve Metadata regressions on the updated PR03 base
1ed86e1 
Keep the ApiVersions wire fallback assertion limited to its retry range,
while successful negotiation verifies both registered API ranges.

Cover Metadata v0-v11 handling of native gateway visibility filtering in
named queries, all-topic discovery, and missing-database retries. Verify
Metadata v12 is rejected before invoking the gateway.

Retain the existing qualified-topic discovery contract on prerequisite
c508de6. New Core-dependent aliases and
persisted table enablement remain deferred.

Validation: Java 11 reactor verify passed 105 unit tests and 5 integration
tests, with Spotless, Checkstyle, RAT, and git diff --check.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 0/0
AI-Contributed/UT: 71/71
@gyang94 @codex
[test] Wait for coordinator metadata before snapshot lease retry
2c7e03d 
Wait for all server gateways to advertise the restarted coordinator before
acquire, release and drop lease checks. Assert that the client still holds
its cached address so each operation exercises metadata refresh and retry.

Rebased PR04 onto PR03 at 86d2735.
Validation: restart case passed in the full Admin suite and focused rerun;
Kafka 110 tests, RPC retry 6 tests, Spotless and Checkstyle passed. The full
Admin suite passed 46/47; the unchanged rolling-upgrade health test first
observed RED instead of GREEN and passed the subsequent focused rerun.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 0/0
AI-Contributed/UT: 11/11
@gyang94 @codex
[kafka] Apply persisted topic enablement to Metadata
a23937d 
Use the actual server default database and the PR03 shared mapper/schema
contract. Resolve aliases by TablePath, preserve requested names, and retain
valid aliases when a missing database triggers metadata discovery retry.

Expose only tables with persisted table.kafka.enabled. Return unknown-topic
for disabled or missing tables and invalid-topic for damaged enabled
contracts in named and all-topic requests. Discovery emits canonical names
without duplicate aliases and retains native authorization and routing.

Verify configuration injection, no fallback lookup or database creation,
enablement/reset changes, invalid historical contracts and v0-v11 behavior
through real Kafka listeners. Preserve the coordinator snapshot lease fix.

Validation: Java 11 clean reactor verify passed 160 distinct cases: 118 Kafka
(including 6 Metadata integration cases), 31 shared schema, 3 atomic DDL,
6 RPC retry and 2 snapshot lease cases. Spotless, Checkstyle and RAT passed.
The four existing PR04 commits were rebased unchanged onto PR03
4a9aa6b and checked with range-diff.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 155/155
AI-Contributed/UT: 571/571
@gyang94 @codex
[kafka] Give topic aliases distinct stable wire IDs
993d234 
Keep canonical topic IDs unchanged and encode alias wire IDs in a separate
namespace using the same Fluss table ID. Kafka clients can then cache both
names and detect table recreation even when the new leader epoch is lower.

Cover Kafka Metadata v10/v11 full and partial updates, recreation from epoch
5 to 0 without a missing-topic response, and native Metadata responses.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 31/31
AI-Contributed/UT: 135/135
@gyang94 @codex
[kafka] Validate Metadata against the updated PR03 contract
dcec633 
Adapt the decoder regression to the configured request handler. Cover
optional three-field metadata columns and legacy offset rejection across
Metadata v0-v11, including native DDL and both real TabletServer listeners.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 0/0
AI-Contributed/UT: 109/109
@gyang94 @codex
[kafka] Simplify Metadata wiring and regression tests
5275f90 
Pass one RPC service through the request handler and validate its
TabletServerGateway capability inside the Metadata backend.

Consolidate overlapping Metadata tests while retaining protocol-version,
alias identity, client cache, authorization and discovery regressions.
Align the custom default database test with PR03 startup creation and
verify that Metadata does not recreate a deleted default database.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 35/35
AI-Contributed/UT: 546/546
@gyang94 @codex
[kafka] Adapt Metadata tests to merged DDL configuration
ffdefa3 
Use KafkaValueConversionType for typed table properties after rebasing
onto main. Preserve the raw invalid value in the disabled-table regression
to keep verifying that Metadata filters it before schema resolution.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 0/0
AI-Contributed/UT: 21/21
@gyang94 @codex
[kafka] Rename default database topic ID namespace
4c70505 
Clarify the namespace name and comment for default-database aliases.
Keep the UUID namespace value and mapping behavior unchanged.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 8/8
AI-Contributed/UT: 0/0
@gyang94 @codex
[kafka] Add independently testable Produce protocol handling
d20cf7c 
Validate non-idempotent Produce v3-v11 and isolate invalid partitions before invoking a protocol-independent backend. Preserve partition order, acknowledgements and owned record data. Leave production registration to the append integration PR.

Validation: Java 11, mvn -o -pl fluss-kafka spotless:apply clean verify; 69 unit tests and 2 integration tests passed, including Checkstyle, Spotless and RAT.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 621/621
AI-Contributed/UT: 366/366
@gyang94 @codex
[kafka] Validate Produce batch offsets and stream compressed records
c568cd5 
Read compressed records through a closeable streaming iterator to avoid
preallocating a record list from the declared batch count. Reject incoming
Kafka batches whose offset range disagrees with their record count before
copying records or invoking the backend. Keep Fluss-assigned offsets independent
of this input validation and preserve partition error isolation.

Cover all five codecs, owned record copies, malformed counts and offset ranges,
and backend-assigned offsets. Java 11 validation passed: 117 unit tests and
4 integration tests, plus Checkstyle, Spotless, RAT, and git diff --check.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 42/42
AI-Contributed/UT: 168/168
@gyang94 @codex
[kafka] Preserve batch timestamp types in Produce commands
91ef63c 
Carry timestamp provenance through a protocol-independent enum on each copied
record. Read the type from its batch for compressed and uncompressed records;
keep timestamp presence separate. Retain the existing constructor with an
explicit UNKNOWN type instead of guessing provenance from the numeric value.

Cover mixed batch types across partitions, absent timestamps, owned data and
legacy constructors. Keep partition validation and acks=0 behavior, and check
that ApiVersions advertises the registered Produce range. Metadata row storage
and production append registration remain outside this protocol-only change.

Validated with Java 11 and this worktree's clean reactor: 152 unit tests and
5 integration tests passed, including ApiVersions and Metadata regressions.
Spotless, Checkstyle, RAT and git diff --check passed.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 64/64
AI-Contributed/UT: 326/326
@gyang94 @codex
[kafka] Verify Produce topic aliases preserve request names
a565b8a 
Document that the protocol backend receives original topic names and returns
results keyed by those names, leaving database resolution to the adapter.
Cover bare and qualified names together across Produce v3-v11, reordered
backend results, local validation failures, backend errors and invalid acks.
Keep null and empty record values distinct in the owned protocol command.

Validated on Java 11 with this worktree's clean reactor: 162 Kafka unit tests,
23 integration tests, and 34 Common/Server schema and DDL tests passed.
Spotless, Checkstyle, RAT and git diff --check passed.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 11/11
AI-Contributed/UT: 150/150
@gyang94 @codex
[kafka] Transcode mapped raw and string records to Arrow
28e43ef 
Reuse the DDL contract to assemble key, value, timestamp and ordered headers into physical rows. Decode strings with strict UTF-8 validation and preserve nullable values. Return owned heap log bytes and recycle Arrow writers on success and failure.

Validation: Java 11, mvn -o -pl fluss-kafka spotless:apply clean verify; 75 unit tests and 2 integration tests passed, including Checkstyle, Spotless and RAT.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 314/314
AI-Contributed/UT: 326/326
@gyang94 @codex
[kafka] Stream record transcoding and reuse owned payloads
3762f51 
Decode and append one record at a time instead of retaining an entire row
list. Borrow command-owned key, value and header bytes during synchronous
conversion while preserving defensive-copy accessors.

Cover row and payload reuse, batches beyond initial Arrow capacity, and
resource cleanup after partial conversion failures.

Validation on Java 11: reactor clean verify passed with 130 unit tests and
4 integration tests, plus Checkstyle, Spotless, RAT and git diff --check.
An additional 25 boundary scenarios passed.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 101/101
AI-Contributed/UT: 224/224
@gyang94 @codex
[kafka] Encode timestamp metadata as UTC TIMESTAMP values
ea5dc34 
Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 13/13
AI-Contributed/UT: 75/75
@gyang94 @codex
[kafka] Transcode binary and string into fixed system columns
8b0a9c6 
Use the shared KafkaTableSchema contract for the fixed value column, raw
optional keys, and last-value-wins header maps. Validate final map and key
nullability, keep optional rescue values null, and remove prototype field
projections and timestamp-column mapping. Reject metadata until native
per-row offsets are available; JSON decoding remains a later stage.

Preserve synchronous streaming, borrowed payloads, owned Arrow output and
failure-path cleanup. Cover all optional system-column combinations, raw
bytes, strict UTF-8, header deduplication, multi-page output and partial
conversion failures without changing null-value filtering.

Validation: Java 11 current-worktree clean reactor, 277 unit tests and 27
integration tests passed, including cumulative Kafka, shared schema, DDL
and Arrow coverage. The final 23 transcoder cases passed again after
strengthening the multi-page test. Spotless, Checkstyle, RAT and diff checks
passed. All SNAPSHOT test dependencies came from this worktree.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 95/95
AI-Contributed/UT: 741/741
@gyang94 @codex
[kafka] Preserve metadata rejection on the updated PR05 base
706ea4d 
The shared schema now accepts three-field Kafka metadata without offsets.
Reject it explicitly during row assembler construction so binary/string
conversion cannot silently emit null metadata. Keep the existing streaming
and borrowed-payload paths unchanged.

Cover nullable and non-null metadata in both formats, and update the legacy
offset-field regression to the current schema validation message.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 5/5
AI-Contributed/UT: 41/41
@gyang94 @codex
[kafka] Transcode three-field metadata from Produce records
dfa5746 
Populate TIMESTAMP_LTZ(3), Kafka batch timestamp type and the requested
partition before Arrow encoding. Preserve timestamp provenance independently
of timestamp presence, and enforce nested timestamp NOT NULL constraints.

Keep synchronous streaming and borrowed key/value/header payloads. Cover all
16 optional-column combinations, timestamp boundaries, partition isolation,
Arrow vector growth and failure atomicity for binary and string values.

Validation: Java 11 offline clean verify; 348 tests passed (317 unit and
31 integration), with Spotless, Checkstyle and RAT checks passing.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 74/74
AI-Contributed/UT: 233/233
@gyang94 @codex
[kafka] Align Produce transcoding with current schema contract
6985a0d 
Use the typed conversion API from the current metadata contract and align
transcoder tests with schema-time nullability validation.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6.1-sol
AI-Contributed/Feature: 16/16
AI-Contributed/UT: 237/237
@gyang94
gyang94 force-pushed the codex/kafka-raw-string-transcode branch from ef27192 to 6985a0d Compare October 19, 2026 04:00
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

[kafka] Transcode mapped raw and string records to Arrow

1 participant

@gyang94