Skip to content

[fix][client] Commit ByteBufPair frames as a single outbound pipeline entry #26459

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
nodece wants to merge 1 commit into from
+206 −14
Closed

[fix][client] Commit ByteBufPair frames as a single outbound pipeline entry #26459

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 29 additions & 14 deletions pulsar-common/src/main/java/org/apache/pulsar/common/protocol/ByteBufPair.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
}
Expand All @@ -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);
}
Expand Down
177 changes: 177 additions & 0 deletions pulsar-common/src/test/java/org/apache/pulsar/common/protocol/ByteBufPairTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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<String> 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<Unpooled> 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;
}
}
Loading