[kafka] Integrate Produce with native Fluss append

[gyang94]

Purpose

Closes #4283. Part of #4185.

Connect Kafka Produce v3-v11 to the TabletServer append gateway for pre-created raw/string tables addressed as database.table. Preserve partition-level error isolation, native acknowledgements and offsets, and immediate delayed-fetch wakeups after append.

Reuse parsed table metadata and compiled write plans, and move Produce metadata lookup, conversion and native submission onto dedicated bounded workers. A slow ZooKeeper lookup no longer occupies the shared RPC request processor for this Produce path.

This is PR07, depending on #4285 (PR06) and independent core PR #3412.

Brief change log

• Apply the shared qualified-topic and table schema contract, retaining the resolved TablePath through lookup and conversion. Invalid topics, missing tables, invalid buckets and conversion failures remain isolated from valid writes.
• Cache up to 1000 parsed TableInfo snapshots and 1000 immutable write plans. Every topic write still refreshes native metadata and authorization; changes to the response, including table identity, schema, properties or bucket-count epoch, invalidate reuse.
• Execute Produce lookup, conversion and synchronous native append through bounded FIFO lanes keyed by the originating RPC worker, preserving same-connection submission order. Workers do not wait for asynchronous acknowledgements. Queue rejection returns REQUEST_TIMED_OUT without running rejected work on the RPC caller.
• Use a separate worker pool sized from netty.server.num-worker-threads and netty.server.max-queued-requests, with queue capacity capped at 1024. Defaults are 8 workers and 500 total queued requests.
• Drain deferred actions immediately after synchronous produceLog invocation, including throws, so acks=all can make replication progress.
• Add the minimal NetworkProtocolPlugin.closeAsync and NettyServer lifecycle support needed to close plugin workers after transport and RPC processors stop, complete queued requests, and handle repeated or pre-start shutdown.

Incremental scope relative to the combined prerequisite baseline: 19 files changed, 2716 insertions(+), 30 deletions(-). This stage changes fluss-kafka plus the protocol lifecycle implementation and tests in fluss-rpc.

Tests

Java 11 reactor verify passed for the source committed at 972c726bf6c774805b70f06686abeb4c62f17962:

• 160 Kafka unit tests, 8 Kafka integration tests and 5 NettyServerTest cases; no failures, errors or skipped tests.
• Checkstyle, Spotless, RAT and git diff --check passed. Committed source files were verified byte-for-byte against the validated working tree.
• Three-TabletServer tests cover native DDL, Kafka discovery, gzip Produce, raw/string, acks=0/1/all, offsets, timestamps/headers and native Fluss readback, including same-named tables in different databases.
• A pipelined producer sends 30 batches of alternating sizes and verifies both offsets and native readback order.
• Unit tests cover concurrent cache reuse, property/epoch changes, authorization revocation, bounded queue rejection, shutdown interruption and cancellation, pending acknowledgements, and immediate deferred-action draining.

The full repository suite, Flink SQL runtime, throughput benchmarks and memory-pressure tests were not run.

API and Format

Activates non-idempotent Produce v3-v11 over the existing Fluss append RPC and Arrow format. Kafka CreateTopics/DeleteTopics, automatic table creation, JSON, authentication, idempotence and transactions remain outside this stage.

ZooKeeper query-count reduction and byte-based admission control are deferred follow-ups. The caches avoid repeated parsing and plan construction; they do not avoid the fresh lookup. The conversion queue is bounded by request count, not retained payload bytes.

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

The comparison base 2a1622da4e18a6659a4b8eb22dba01c927ff279a combines PR06 at 04a9093b331cfb1835493627ef0ab0adb969b8c2 with the existing core PR #3412 at 969f985abbad907e99588a1d0f3252196e2fdee8. The delayed-fetch changes remain owned and reviewed in #3412. PR07 additionally includes the small RPC protocol shutdown hook required for its executor. After prerequisites merge, update the baseline and validate again.

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