From 6f97f56ebb19b1b4f2c391654015f763dd85706a Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Fri, 4 Sep 2026 15:20:19 +0800 Subject: [PATCH] [fix][client] Commit ByteBufPair frames as a single outbound pipeline 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). --- .../pulsar/common/protocol/ByteBufPair.java | 43 +++-- .../common/protocol/ByteBufPairTest.java | 177 ++++++++++++++++++ 2 files changed, 206 insertions(+), 14 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/ByteBufPair.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/ByteBufPair.java index 5dbf07f8e835b..90dc77db2e554 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/ByteBufPair.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/ByteBufPair.java @@ -20,6 +20,7 @@ import com.google.common.annotations.VisibleForTesting; import io.netty.buffer.ByteBuf; +import io.netty.buffer.CompositeByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandler.Sharable; import io.netty.channel.ChannelHandlerContext; @@ -152,16 +153,25 @@ public static ChannelOutboundHandlerAdapter getEncoder(boolean tlsEnabled) { @SuppressWarnings("checkstyle:JavadocType") public static class Encoder extends ChannelOutboundHandlerAdapter { @Override - public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { if (msg instanceof ByteBufPair) { ByteBufPair b = (ByteBufPair) msg; - - // Write each buffer individually on the socket. The retain() here is needed to preserve the fact that - // ByteBuf are automatically released after a write. If the ByteBufPair ref count is increased and it - // gets written multiple times, the individual buffers refcount should be reflected as well. + CompositeByteBuf frame = null; try { - ctx.write(b.getFirst().retainedDuplicate(), ctx.voidPromise()); - ctx.write(b.getSecond().retainedDuplicate(), promise); + ByteBuf first = b.getFirst(); + ByteBuf second = b.getSecond(); + // Commit the whole frame as a single pipeline entry: two separate writes allowed + // a failure in between to enqueue only the header half and desynchronize the peer's + // frame parser. The retain() gives the composite independent component claims, so + // writing a pair the caller has retained again (resend) keeps working. + frame = Unpooled.compositeBuffer(2); + frame.addComponent(true, first.retain()); + frame.addComponent(true, second.retain()); + ctx.write(frame, promise); + frame = null; + } catch (Throwable t) { + ReferenceCountUtil.safeRelease(frame); + promise.tryFailure(t); } finally { ReferenceCountUtil.safeRelease(b); } @@ -175,16 +185,21 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) @SuppressWarnings("checkstyle:JavadocType") public static class CopyingEncoder extends ChannelOutboundHandlerAdapter { @Override - public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { if (msg instanceof ByteBufPair) { ByteBufPair b = (ByteBufPair) msg; - - // Some handlers in the pipeline will modify the bytebufs passed in to them (i.e. SslHandler). - // For these handlers, we need to pass a copy of the buffers as the source buffers may be cached - // for multiple requests. + CompositeByteBuf frame = null; try { - ctx.write(b.getFirst().copy(), ctx.voidPromise()); - ctx.write(b.getSecond().copy(), promise); + ByteBuf first = b.getFirst(); + ByteBuf second = b.getSecond(); + frame = Unpooled.compositeBuffer(2); + frame.addComponent(true, first.copy()); + frame.addComponent(true, second.copy()); + ctx.write(frame, promise); + frame = null; + } catch (Throwable t) { + ReferenceCountUtil.safeRelease(frame); + promise.tryFailure(t); } finally { ReferenceCountUtil.safeRelease(b); } diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/ByteBufPairTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/ByteBufPairTest.java index ab54df94fcdcc..b3f02a12907cc 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/ByteBufPairTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/ByteBufPairTest.java @@ -20,13 +20,29 @@ import static org.mockito.Mockito.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertTrue; import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBufUtil; import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.DefaultChannelPromise; +import io.netty.channel.VoidChannelPromise; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.util.IllegalReferenceCountException; +import io.netty.util.concurrent.ImmediateEventExecutor; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; import org.apache.pulsar.common.allocator.PulsarByteBufAllocator; +import org.mockito.MockedStatic; import org.testng.annotations.Test; public class ByteBufPairTest { @@ -94,4 +110,165 @@ public void testCoalesce() { assertEquals(new String(ByteBufUtil.getBytes(coalesced)), "helloworld"); coalesced.release(); } + + /** + * A frame must reach the outbound pipeline as a single message, not as a + * header entry followed by a payload entry. + */ + @Test + public void testEncoderCommitsFrameAsSingleMessage() { + EmbeddedChannel channel = new EmbeddedChannel(ByteBufPair.ENCODER); + ByteBuf b1 = Unpooled.wrappedBuffer("hello".getBytes()); + ByteBuf b2 = Unpooled.wrappedBuffer("world".getBytes()); + ByteBufPair buf = ByteBufPair.get(b1, b2); + + channel.writeOutbound(buf); + + ByteBuf frame = channel.readOutbound(); + assertNotNull(frame); + assertEquals(frame.readableBytes(), 10); + assertEquals(new String(ByteBufUtil.getBytes(frame)), "helloworld"); + assertNull(channel.readOutbound()); + frame.release(); + channel.finishAndReleaseAll(); + } + + /** Same as above for the copying encoder used on the TLS path. */ + @Test + public void testCopyingEncoderCommitsFrameAsSingleMessage() { + EmbeddedChannel channel = new EmbeddedChannel(ByteBufPair.COPYING_ENCODER); + ByteBuf b1 = Unpooled.wrappedBuffer("hello".getBytes()); + ByteBuf b2 = Unpooled.wrappedBuffer("world".getBytes()); + ByteBufPair buf = ByteBufPair.get(b1, b2); + + channel.writeOutbound(buf); + + ByteBuf frame = channel.readOutbound(); + assertNotNull(frame); + assertEquals(new String(ByteBufUtil.getBytes(frame)), "helloworld"); + assertNull(channel.readOutbound()); + assertEquals(buf.refCnt(), 0); + assertEquals(b1.refCnt(), 0); + assertEquals(b2.refCnt(), 0); + frame.release(); + channel.finishAndReleaseAll(); + } + + /** + * The same pair can be written multiple times, as a resend after reconnect does. + */ + @Test + public void testEncoderSupportsMultipleWritesOfSamePair() { + EmbeddedChannel channelA = new EmbeddedChannel(ByteBufPair.ENCODER); + EmbeddedChannel channelB = new EmbeddedChannel(ByteBufPair.ENCODER); + ByteBuf b1 = Unpooled.wrappedBuffer("hello".getBytes()); + ByteBuf b2 = Unpooled.wrappedBuffer("world".getBytes()); + ByteBufPair buf = ByteBufPair.get(b1, b2); + + // Each write takes its own pair claim first, as ProducerImpl does (op.cmd.retain()). + buf.retain(); + channelA.writeOutbound(buf); + buf.retain(); + channelB.writeOutbound(buf); + + ByteBuf frameA = channelA.readOutbound(); + ByteBuf frameB = channelB.readOutbound(); + assertEquals(new String(ByteBufUtil.getBytes(frameA)), "helloworld"); + assertEquals(new String(ByteBufUtil.getBytes(frameB)), "helloworld"); + frameA.release(); + frameB.release(); + + assertEquals(buf.refCnt(), 1); + buf.release(); + assertEquals(buf.refCnt(), 0); + assertEquals(b1.refCnt(), 0); + assertEquals(b2.refCnt(), 0); + channelA.finishAndReleaseAll(); + channelB.finishAndReleaseAll(); + } + + /** + * If one pair's component is concurrently released (as the send-timeout disposal can do + * during a reconnect window), every entry that still reaches the wire must be a complete + * frame — a header-only entry would desynchronize every frame after it. + */ + @Test + public void testEncoderFailedBuildKeepsStreamAligned() { + EmbeddedChannel channel = new EmbeddedChannel(ByteBufPair.ENCODER); + ByteBufPair good = ByteBufPair.get(Unpooled.wrappedBuffer("good1".getBytes()), + Unpooled.wrappedBuffer("good2".getBytes())); + ByteBuf bad2 = Unpooled.wrappedBuffer("payload".getBytes()); + ByteBufPair bad = ByteBufPair.get(Unpooled.wrappedBuffer("head!".getBytes()), bad2); + ByteBufPair good3 = ByteBufPair.get(Unpooled.wrappedBuffer("good3".getBytes()), + Unpooled.wrappedBuffer("good4".getBytes())); + bad2.release(); + + for (ByteBufPair p : Arrays.asList(good, bad, good3)) { + try { + channel.writeOutbound(p); + } catch (IllegalReferenceCountException acceptable) { + // a failed write is acceptable here, a partial entry is not + } + } + + List frames = new ArrayList<>(); + ByteBuf frame; + while ((frame = channel.readOutbound()) != null) { + frames.add(new String(ByteBufUtil.getBytes(frame))); + frame.release(); + } + assertEquals(frames, Arrays.asList("good1good2", "good3good4")); + channel.finishAndReleaseAll(); + } + + @Test + public void testEncoderFailedBuildCommitsNothing() { + ByteBuf bad1 = Unpooled.wrappedBuffer("head!".getBytes()); + ByteBuf bad2 = Unpooled.wrappedBuffer("payload".getBytes()); + ByteBufPair bad = ByteBufPair.get(bad1, bad2); + // Simulate the concurrent release of the second component before the encoder takes its claim. + bad2.release(); + assertEquals(bad2.refCnt(), 0); + + ChannelHandlerContext ctx = mock(ChannelHandlerContext.class); + DefaultChannelPromise promise = new DefaultChannelPromise( + channelMock(), ImmediateEventExecutor.INSTANCE); + + bad.retain(); + ByteBufPair.ENCODER.write(ctx, bad, promise); + + verify(ctx, never()).write(any(), any()); + assertTrue(promise.isDone() && !promise.isSuccess(), "the write promise must be failed"); + // the composite returned the first component's claim; the finally block consumed the write claim + assertEquals(bad1.refCnt(), 1); + assertEquals(bad.refCnt(), 1); + } + + @Test + public void testEncoderAllocationFailureFailsWriteAndReleasesPair() { + ByteBuf b1 = Unpooled.wrappedBuffer("hello".getBytes()); + ByteBuf b2 = Unpooled.wrappedBuffer("world".getBytes()); + ByteBufPair buf = ByteBufPair.get(b1, b2); + ChannelHandlerContext ctx = mock(ChannelHandlerContext.class); + DefaultChannelPromise promise = new DefaultChannelPromise( + channelMock(), ImmediateEventExecutor.INSTANCE); + + try (MockedStatic pooled = mockStatic(Unpooled.class)) { + pooled.when(() -> Unpooled.compositeBuffer(2)).thenThrow(new OutOfMemoryError("test")); + ByteBufPair.ENCODER.write(ctx, buf, promise); + } + + verify(ctx, never()).write(any(), any()); + assertTrue(promise.isDone() && !promise.isSuccess(), "the write promise must be failed"); + assertEquals(buf.refCnt(), 0); + assertEquals(b1.refCnt(), 0); + assertEquals(b2.refCnt(), 0); + } + + private static Channel channelMock() { + Channel channel = mock(Channel.class); + when(channel.eventLoop()).thenReturn(new EmbeddedChannel().eventLoop()); + when(channel.voidPromise()).thenReturn(new VoidChannelPromise(channel, true)); + return channel; + } }