[fix][client] Commit ByteBufPair frames as a single outbound pipeline entry - #26459
[fix][client] Commit ByteBufPair frames as a single outbound pipeline entry#26459nodece wants to merge 1 commit into
Conversation
… entry Motivation ByteBufPair.Encoder wrote each frame as two independent pipeline writes: the header half with the void promise, then the payload half with the real promise. The two entries had no common failure domain: if claiming or writing the payload half failed after the header half was already enqueued (for example when a concurrent send-timeout disposal released the pair's buffers between the two writes), a header-only entry reached the outbound buffer and permanently desynchronized the peer's frame stream. The broker then rejects everything that follows with "Failed to verify checksum" / TooLongFrameException and closes the connection. Because the failure depends on a race between the timeout disposal and the write path, it surfaced only under memory pressure and connection churn, not in steady state. Modifications - Encoder and CopyingEncoder now build a single CompositeByteBuf per frame (retain-based component ownership for Encoder, copies for CopyingEncoder) and commit it with one ctx.write() carrying the real promise: a frame is handed to the outbound buffer whole or not at all, and on a failed build the write promise is failed and nothing is committed. - The pair keeps its original component references, so a pair written multiple times (resend after reconnect) still works. - Add regression tests: single-message commit, repeated writes of the same pair, and stream alignment when a component is concurrently released (no partial entry may reach the wire).
|
@nodece If the first write fails, the second must fail too. Add a composite bytebuf here does not make much sense. ByteBufPair was created precisely to avoid the overhead of the general purpose composite bytebuf. |
It seems that Netty is used in the wrong way if this would happen. Concurrently releasing something that is already queued should be addressed instead. It could multiple other issues if already queued buffers are released. When |
|
@merlimat @lhotari thanks for the review — I re-validated today and you were both right. I tested the exact head of #26456 alone (bytecode-verified: the two-write encoder untouched, only the release-side fix in), in the same harness where unmodified master corrupts the broker within the first round ( My earlier conclusion that the encoder also needed a change came from a run whose jar was built from a dirty working tree; it doesn't reproduce with the isolated commit. That's on me. Closing this PR. #26456 is the right fix — it's exactly the "address the concurrent release at the source" that @lhotari described, and @merlimat's point about the ByteBufPair design intent stands. |
5 participants
Motivation
ByteBufPair.Encoder— the outbound encoder that both the client and brokerPulsarChannelInitializerinstall for the binary protocol — serialized each frame as two independent pipeline writes: the header half with the void promise, then the payload half carrying the real write promise. The two entries share no failure domain.If anything fails after the header half is enqueued but before the payload half is claimed — most notably when the send-timeout path concurrently releases the pair's buffers while the frame is being written on the event loop — a header-only entry is committed to the channel outbound buffer and reaches the wire. (In
ProducerImpl, thecnx == nullreconnect-window branch offailPendingMessagesdisposes in-flight ops on the timer thread, which can race the write of the same op on the connection's event loop.)The consequence is severe: the peer's
LengthFieldBasedFrameDecoderpermanently loses frame sync on that connection, and the broker rejects everything that follows:(
1515870814is0x5A5A5A5A— the broker read message payload bytes as the frame length, i.e. the stream was already misaligned mid-payload.) The connection is then closed and all in-flight messages on it are lost.This is a long-lived latent defect that only surfaces under a specific conjunction of conditions, which is why intermittent reports with this exact signature have historically been attributed to the network layer or intermediaries (e.g. #21557):
sendTimeout(e.g. 3s);-XX:MaxDirectMemorySizemakesBits.reserveMemoryblock the allocating thread for seconds near the cap — aging messages past the timeout exactly while their writes are queued on the event loop.Modifications
EncoderandCopyingEncodernow build each frame as a singleCompositeByteBuf(component claims taken viaretain()inEncoder, copies inCopyingEncoder) and commit it with onectx.write()carrying the real promise: a frame reaches the outbound buffer whole or not at all.Verifying this change
This change added tests and can be verified as follows:
ByteBufPairTestnow covers:op.cmd.retain());testEncoderFailedBuildKeepsStreamAlignedis the minimal deterministic repro: on the previous two-write encoder it fails with a header-only"head!"entry committed between the good frames;End-to-end reproduction (how the defect was confirmed and the patch validated):
batchingMaxMessages=1000,batchingMaxBytes=128KB,batchingMaxPublishDelay=1ms,sendTimeout(3, SECONDS), no compression, 4 threads x 8KB payloads, client JVM-XX:MaxDirectMemorySize=48m, connecting through the proxy;docker restartthe proxy during active traffic (twice per 50s round);Failed to verify checksum/TooLongFrameException/unknown tag type.Results: the unmodified
masterclient corrupted the stream in round 2 of this harness with the exact signature quoted above; with this patch, 34 rounds (~770k messages) produced zero corrupt entries — the client-side send errors after each restart were transient and always recovered.Does this pull request potentially affect one of the following parts:
Documentation
no-docneeded (internal fix, no user-facing behavior or configuration change)