Skip to content

[kafka] Handle basic Produce requests independently of storage - #4284

Open
gyang94 wants to merge 14 commits into
apache:mainfrom
gyang94:codex/kafka-produce-protocol
Open

[kafka] Handle basic Produce requests independently of storage#4284
gyang94 wants to merge 14 commits into
apache:mainfrom
gyang94:codex/kafka-produce-protocol

Conversation

@gyang94

@gyang94 gyang94 commented Sep 10, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Purpose

Closes #4281. Part of #4185.

Implement non-idempotent Produce v3-v11 through an independently testable handler and a narrow backend contract. Invalid partitions receive their own errors while valid partitions continue to the backend; responses preserve the original request topic names and partition order.

This is PR05, depending on #4261 (PR04 Metadata), based on ecbb368832efd3d5f542880c190e419f3ad5604b.

Brief change log

  • Validate acknowledgements, topic/partition identities, batch checksums and incoming batch offset/count consistency. Reject transactional, idempotent and control batches.
  • Stream compressed records with a closeable iterator, then copy keys, values and headers into immutable commands whose lifetime is independent of the request buffer. Preserve null and empty values distinctly.
  • Preserve each record's timestamp and its source batch timestamp type, with timestamp presence tracked separately. Leave native storage offsets to the append backend.
  • Preserve bare and fully qualified topic names across backend commands, reordered results, validation failures and backend failures. Keep partition errors isolated.
  • Verify Produce capability advertisement in a test registry and verify that acks=0 sends no response and releases the request buffer on success and failure.

Incremental scope relative to PR04: 7 files, +1712 / -0. Production Produce registration, Row conversion and native append are delivered by later PRs.

Tests

Java 11 at 0dde872227c68deb96e5959192af7f801fed9861:

  • Reactor verify with -pl fluss-kafka -am and all Kafka module test classes: 160 unit tests and 19 integration tests passed, with no failures, errors or skips.
  • Coverage includes Produce v3-v11, all five compression codecs, malformed batches, timestamp types, topic aliases, acks=0 and real-cluster Metadata regression tests.
  • The custom default database regression verifies startup creation and verifies that Metadata does not recreate a deleted database.
  • Spotless, Checkstyle, RAT and git diff --check passed. Reactor dependencies were used without external Fluss SNAPSHOT artifacts on the Kafka test classpath.

The full repository test suite and Flink SQL runtime were not run. Real Produce append semantics remain part of the later integration PRs.

API and Format

Adds independently testable non-idempotent Produce v3-v11 handling. The production endpoint continues to register ApiVersions and Metadata at this stage.

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 PR05 alone.

Generative AI disclosure: Codex assisted with 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-produce-protocol branch from 936fd3f to ff27c55 Compare September 17, 2026 09:04
@gyang94
gyang94 force-pushed the codex/kafka-produce-protocol branch 2 times, most recently from fec411d to 0dde872 Compare October 14, 2026 14:04
@gyang94
gyang94 marked this pull request as ready for review October 14, 2026 14:25
gyang94 and others added 14 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
gyang94 force-pushed the codex/kafka-produce-protocol branch from 0dde872 to a565b8a Compare October 19, 2026 03:48
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] Handle basic Produce requests independently of storage

1 participant

@gyang94