[dao-jun]

Purpose

KafkaCommandDecoder never registered its channels with the RequestChannel that serves them, so backpressure on the request queue could not pause reads on Kafka listener connections: under Kafka-side overload the unbounded per-processor request queue kept growing while queued Kafka produce requests retained their borrowed (potentially large) buffers, and connections feeding the queue were invisible to the pause/resume mechanism that already protects Fluss-protocol connections (NettyServerHandler registers/unregisters its channels). This wires Kafka listener channels into the same mechanism.

Linked issue: part of umbrella #4185 (Kafka Protocols Compatibility – Basic Produce), whose Future Work lists bounded Produce admission and flow control. Per the umbrella's review workflow, this PR references it without closing it.

Brief change log

• KafkaCommandDecoder: register the channel in channelActive and unregister it in channelInactive, mirroring NettyServerHandler for the Fluss protocol. Unregistering also removes closed channels from the backpressure bookkeeping so connection churn cannot leak channels (and their pipelines) into the registered-channel set. channelInactive follows NettyServerHandler's try/finally shape so pipeline propagation still happens if cleanup ever throws.
• RequestChannel: add a @VisibleForTesting isChannelRegistered(Channel) read-only probe so tests can pin the unregister half of the lifecycle (without it, a dropped unregisterChannel call is unobservable because pauseAllChannels/resumeAllChannels skip inactive channels anyway).

Tests

New KafkaChannelBackpressureTest (plain EmbeddedChannel, no cluster fixture) covers:

• a request reaching the queue through the real decode → putRequest path pauses the channel's reads when the backpressure threshold is crossed, and draining below the resume threshold resumes them;
• closing a channel unregisters it: a later backpressure episode pauses the still-open channel but not the closed one, and isChannelRegistered no longer reports the closed channel;
• a frame sent while reads are paused is withheld by FlowControlHandler and re-delivered (not lost) once reads resume.

Red/green verified: removing the registerChannel call fails all three tests; removing the unregisterChannel call fails the unregister test. mvn test -pl fluss-rpc,fluss-kafka is green on top of current main.

API and Format

No public API or storage format change; the only main-code addition besides the two lifecycle calls is a @VisibleForTesting accessor on the internal RequestChannel.

Documentation

No new feature or configuration option; no documentation change needed.

Generative AI disclosure:

• No generative AI tools used
• Yes — tool: Claude Code