[dao-jun]

Purpose

Server channels start reading the moment they connect, so a channel that registers while request-queue backpressure is already active is never paused: reconcileBackpressure early-returns while the queue sits between the resume and backpressure thresholds, and pauseAllChannels only runs at the pause transition. Under overload, reconnecting clients (timed-out producers that reconnect into fresh connections) keep feeding the unbounded per-processor request queue — the exact memory growth backpressure exists to prevent.

This changes server channels to start with reads disabled and makes RequestChannel.registerChannel the single point that decides reads, aligned with the current backpressure state.

Linked issue: part of umbrella #4185 (Kafka Protocols Compatibility – Basic Produce), whose Future Work lists bounded Produce admission and flow control; referenced without closing it.

Brief change log

• ServerChannelInitializer: accepted Fluss-protocol channels start with autoRead=false; they cannot read a single byte before their backpressure state has been decided.
• RequestChannel.registerChannel: aligns the channel with the current state — enables reads when backpressure is inactive, pauses the channel when backpressure is active. A channel arriving during an episode therefore joins the paused set and is released by the next resume transition once the queue drains below the resume threshold. Inactive channels are skipped (registration races with disconnects on the event loop, and a closed channel's read state no longer matters).
• Documentation: the registerChannel javadoc previously documented the opposite behavior ("new channels are NOT immediately paused ... can cause deadlock"); it and the backpressureLock javadoc now describe the actual semantics, including why registration needs no lock (the add-before-flag-read ordering plus event-loop serialization of read-state changes).
• RequestChannelTest: new cases for registering during active backpressure, enabling reads for a start-paused channel, and skipping inactive channels; the hand-written TestChannel fake gained an active flag.

Complements #4897 (which registers Kafka listener channels): once both merge, channels of any protocol that feed a queue are gated at registration. The two PRs both touch RequestChannel.java in non-overlapping regions; whichever lands second needs a trivial rebase.

Tests

• RequestChannelTest.testRegisterDuringActiveBackpressureStartsPaused — a channel registering while backpressure is active stays paused and resumes with the rest when the queue drains (red against both the previous code and the intermediate "enable-only" variant).
• RequestChannelTest.testRegisterEnablesReadsForStartPausedChannel — a channel delivered with reads disabled gets reads enabled on registration (red against previous code).
• RequestChannelTest.testRegisterSkipsInactiveChannel — an inactive channel's read state is left untouched (red against the unguarded variant, which also NPEs on test doubles whose config() is null, caught by MessageCodecTest).
• mvn test -pl fluss-rpc green.

API and Format

No public API or storage format change; registerChannel semantics are internal to the server RPC layer.

Documentation

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

Generative AI disclosure:

• No generative AI tools used
• Yes — tool: Claude Code (design co-authored with the contributor)