Skip to content

Commit 94890cb

Browse files
lhotarilhotari
authored
[fix][broker] Restore per-message auto-read flow control broken by Netty 4.2.15 (#26013)
1 parent 85bdf67 commit 94890cb

3 files changed

Lines changed: 365 additions & 2 deletions

File tree

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222
import io.netty.channel.ChannelHandler;
2323
import io.netty.channel.ChannelInitializer;
2424
import io.netty.channel.socket.SocketChannel;
25-
import io.netty.handler.flow.FlowControlHandler;
2625
import io.netty.handler.flush.FlushConsolidationHandler;
2726
import io.netty.handler.ssl.SslHandler;
2827
import java.util.concurrent.TimeUnit;
@@ -36,6 +35,7 @@
3635
import org.apache.pulsar.common.protocol.OptionalProxyProtocolDecoder;
3736
import org.apache.pulsar.common.util.PulsarSslConfiguration;
3837
import org.apache.pulsar.common.util.PulsarSslFactory;
38+
import org.apache.pulsar.common.util.netty.PulsarFlowControlHandler;
3939

4040
@CustomLog
4141
public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel> {
@@ -98,7 +98,10 @@ protected void initChannel(SocketChannel ch) throws Exception {
9898
// as they like for any given input. so, disabling auto-read on `ByteToMessageDecoder` doesn't work properly and
9999
// ServerCnx ends up reading higher number of messages and broker can not throttle the messages by disabling
100100
// auto-read.
101-
ch.pipeline().addLast("flowController", new FlowControlHandler());
101+
// PulsarFlowControlHandler is used instead of Netty's FlowControlHandler since Netty 4.2.15 changed the
102+
// behavior to ignore setAutoRead(false) made by a downstream handler while queued messages are being
103+
// delivered, which breaks throttling of already buffered messages. See PulsarFlowControlHandler javadoc.
104+
ch.pipeline().addLast("flowController", new PulsarFlowControlHandler());
102105
// using "ChannelHandler" type to workaround an IntelliJ bug that shows a false positive error
103106
ChannelHandler cnx = newServerCnx(pulsar, listenerName);
104107
ch.pipeline().addLast("handler", cnx);

‎pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/PulsarFlowControlHandler.java‎

Lines changed: 245 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,245 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.apache.pulsar.common.util.netty;
20+
21+
import io.netty.channel.ChannelConfig;
22+
import io.netty.channel.ChannelDuplexHandler;
23+
import io.netty.channel.ChannelHandler;
24+
import io.netty.channel.ChannelHandlerContext;
25+
import io.netty.handler.codec.ByteToMessageDecoder;
26+
import io.netty.util.Recycler;
27+
import io.netty.util.ReferenceCountUtil;
28+
import io.netty.util.internal.ObjectPool.Handle;
29+
import io.netty.util.internal.logging.InternalLogger;
30+
import io.netty.util.internal.logging.InternalLoggerFactory;
31+
import java.util.ArrayDeque;
32+
import java.util.Queue;
33+
34+
/**
35+
* The {@link PulsarFlowControlHandler} ensures that only one message per {@code read()} is sent downstream.
36+
*
37+
* <p>Classes such as {@link ByteToMessageDecoder} are free to emit as many events as they like for any given input.
38+
* A channel's auto reading configuration doesn't usually apply in these scenarios. This is causing problems in
39+
* downstream {@link ChannelHandler}s that would like to hold subsequent events while they're processing one event.
40+
*
41+
* <p>This class is derived from {@code io.netty.handler.flow.FlowControlHandler} of Netty 4.2.14.Final
42+
* (Copyright 2016 The Netty Project, licensed under the Apache License, version 2.0), which is behaviorally
43+
* identical to the implementation in Netty 4.1.135.Final. Up to Netty
44+
* 4.1.135/4.2.14, the dequeue loop of Netty's handler re-checked {@link ChannelConfig#isAutoRead()} before
45+
* releasing each queued message, so a downstream handler calling {@code setAutoRead(false)} from inside
46+
* {@code channelRead} stopped the delivery of queued messages immediately. Netty 4.2.15 (netty/netty#16837,
47+
* backport of netty/netty#15053; the equivalent 4.1 change is netty/netty#16912) rewrote the handler to decide
48+
* once up front: with auto-read enabled it dequeues the entire queue and ignores auto-read changes made mid-drain
49+
* by downstream handlers. Pulsar's reactive request throttling (see {@code ServerCnxThrottleTracker} in the broker)
50+
* pauses a connection by disabling auto-read while processing a message and relies on the delivery of queued
51+
* messages stopping immediately; with Netty's rewritten handler, the whole queued backlog would be delivered
52+
* regardless, defeating throttling (and rate limiting) of already-buffered traffic. This copy preserves the
53+
* pre-4.2.15 per-message auto-read behavior.
54+
*
55+
* @see ChannelConfig#setAutoRead(boolean)
56+
*/
57+
public class PulsarFlowControlHandler extends ChannelDuplexHandler {
58+
private static final InternalLogger logger = InternalLoggerFactory.getInstance(PulsarFlowControlHandler.class);
59+
60+
private final boolean releaseMessages;
61+
62+
private RecyclableArrayDeque queue;
63+
64+
private ChannelConfig config;
65+
66+
private boolean shouldConsume;
67+
68+
public PulsarFlowControlHandler() {
69+
this(true);
70+
}
71+
72+
public PulsarFlowControlHandler(boolean releaseMessages) {
73+
this.releaseMessages = releaseMessages;
74+
}
75+
76+
/**
77+
* Determine if the underlying {@link Queue} is empty. This method exists for
78+
* testing, debugging and inspection purposes and it is not Thread safe!
79+
*/
80+
boolean isQueueEmpty() {
81+
return queue == null || queue.isEmpty();
82+
}
83+
84+
/**
85+
* Releases all messages and destroys the {@link Queue}.
86+
*/
87+
private void destroy() {
88+
if (queue != null) {
89+
90+
if (!queue.isEmpty()) {
91+
logger.trace("Non-empty queue: {}", queue);
92+
93+
if (releaseMessages) {
94+
Object msg;
95+
while ((msg = queue.poll()) != null) {
96+
ReferenceCountUtil.safeRelease(msg);
97+
}
98+
}
99+
}
100+
101+
queue.recycle();
102+
queue = null;
103+
}
104+
}
105+
106+
@Override
107+
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
108+
config = ctx.channel().config();
109+
}
110+
111+
@Override
112+
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
113+
super.handlerRemoved(ctx);
114+
if (!isQueueEmpty()) {
115+
dequeue(ctx, queue.size());
116+
}
117+
destroy();
118+
}
119+
120+
@Override
121+
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
122+
destroy();
123+
ctx.fireChannelInactive();
124+
}
125+
126+
@Override
127+
public void read(ChannelHandlerContext ctx) throws Exception {
128+
if (dequeue(ctx, 1) == 0) {
129+
// It seems no messages were consumed. We need to read() some
130+
// messages from upstream and once one arrives it need to be
131+
// relayed to downstream to keep the flow going.
132+
shouldConsume = true;
133+
ctx.read();
134+
} else if (config.isAutoRead()) {
135+
ctx.read();
136+
}
137+
}
138+
139+
@Override
140+
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
141+
if (queue == null) {
142+
queue = RecyclableArrayDeque.newInstance();
143+
}
144+
145+
queue.offer(msg);
146+
147+
// We just received one message. Do we need to relay it regardless
148+
// of the auto reading configuration? The answer is yes if this
149+
// method was called as a result of a prior read() call.
150+
int minConsume = shouldConsume ? 1 : 0;
151+
shouldConsume = false;
152+
153+
dequeue(ctx, minConsume);
154+
}
155+
156+
@Override
157+
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
158+
if (isQueueEmpty()) {
159+
ctx.fireChannelReadComplete();
160+
} else {
161+
// Don't relay completion events from upstream as they
162+
// make no sense in this context. See dequeue() where
163+
// a new set of completion events is being produced.
164+
}
165+
}
166+
167+
/**
168+
* Dequeues one or many (or none) messages depending on the channel's auto
169+
* reading state and returns the number of messages that were consumed from
170+
* the internal queue.
171+
*
172+
* <p>The {@code minConsume} argument is used to force {@code dequeue()} into
173+
* consuming that number of messages regardless of the channel's auto
174+
* reading configuration.
175+
*
176+
* @see #read(ChannelHandlerContext)
177+
* @see #channelRead(ChannelHandlerContext, Object)
178+
*/
179+
private int dequeue(ChannelHandlerContext ctx, int minConsume) {
180+
int consumed = 0;
181+
182+
// fireChannelRead(...) may call ctx.read() and so this method may reentrance. Because of this we need to
183+
// check if queue was set to null in the meantime and if so break the loop.
184+
while (queue != null && (consumed < minConsume || config.isAutoRead())) {
185+
Object msg = queue.poll();
186+
if (msg == null) {
187+
break;
188+
}
189+
190+
++consumed;
191+
ctx.fireChannelRead(msg);
192+
}
193+
194+
// We're firing a completion event every time one (or more)
195+
// messages were consumed and the queue ended up being drained
196+
// to an empty state.
197+
if (queue != null && queue.isEmpty()) {
198+
queue.recycle();
199+
queue = null;
200+
201+
if (consumed > 0) {
202+
ctx.fireChannelReadComplete();
203+
}
204+
}
205+
206+
return consumed;
207+
}
208+
209+
/**
210+
* A recyclable {@link ArrayDeque}.
211+
*/
212+
private static final class RecyclableArrayDeque extends ArrayDeque<Object> {
213+
214+
private static final long serialVersionUID = 0L;
215+
216+
/**
217+
* A value of {@code 2} should be a good choice for most scenarios.
218+
*/
219+
private static final int DEFAULT_NUM_ELEMENTS = 2;
220+
221+
private static final Recycler<RecyclableArrayDeque> RECYCLER =
222+
new Recycler<RecyclableArrayDeque>() {
223+
@Override
224+
protected RecyclableArrayDeque newObject(Handle<RecyclableArrayDeque> handle) {
225+
return new RecyclableArrayDeque(DEFAULT_NUM_ELEMENTS, handle);
226+
}
227+
};
228+
229+
public static RecyclableArrayDeque newInstance() {
230+
return RECYCLER.get();
231+
}
232+
233+
private final Handle<RecyclableArrayDeque> handle;
234+
235+
private RecyclableArrayDeque(int numElements, Handle<RecyclableArrayDeque> handle) {
236+
super(numElements);
237+
this.handle = handle;
238+
}
239+
240+
public void recycle() {
241+
clear();
242+
handle.recycle(this);
243+
}
244+
}
245+
}

‎pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/PulsarFlowControlHandlerTest.java‎

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.apache.pulsar.common.util.netty;
20+
21+
import static org.testng.Assert.assertEquals;
22+
import static org.testng.Assert.assertFalse;
23+
import static org.testng.Assert.assertTrue;
24+
import io.netty.channel.ChannelHandlerContext;
25+
import io.netty.channel.ChannelInboundHandlerAdapter;
26+
import io.netty.channel.embedded.EmbeddedChannel;
27+
import java.util.ArrayList;
28+
import java.util.List;
29+
import org.testng.annotations.Test;
30+
31+
/**
32+
* Tests the auto-read behavior of {@link PulsarFlowControlHandler} that Pulsar's reactive request throttling
33+
* depends on: a downstream handler disabling auto-read from inside {@code channelRead} must stop the delivery
34+
* of queued messages immediately. Netty's {@code FlowControlHandler} lost this behavior in Netty 4.2.15
35+
* (netty/netty#16837), which is why Pulsar carries this copy of the previous implementation.
36+
*/
37+
public class PulsarFlowControlHandlerTest {
38+
39+
/**
40+
* Downstream handler that records received messages and disables auto-read whenever the number of received
41+
* messages reaches the next configured threshold, mimicking how ServerCnxThrottleTracker pauses a connection
42+
* while processing a message.
43+
*/
44+
private static class ThrottlingHandler extends ChannelInboundHandlerAdapter {
45+
private final List<Object> received = new ArrayList<>();
46+
private int pauseAtCount;
47+
48+
ThrottlingHandler(int pauseAtCount) {
49+
this.pauseAtCount = pauseAtCount;
50+
}
51+
52+
void pauseAgainAtCount(int count) {
53+
this.pauseAtCount = count;
54+
}
55+
56+
@Override
57+
public void channelRead(ChannelHandlerContext ctx, Object msg) {
58+
received.add(msg);
59+
if (received.size() == pauseAtCount) {
60+
ctx.channel().config().setAutoRead(false);
61+
}
62+
}
63+
}
64+
65+
@Test
66+
public void shouldStopDeliveryWhenAutoReadIsDisabledDuringChannelRead() {
67+
PulsarFlowControlHandler flowControlHandler = new PulsarFlowControlHandler();
68+
ThrottlingHandler throttlingHandler = new ThrottlingHandler(1);
69+
EmbeddedChannel channel = new EmbeddedChannel(flowControlHandler, throttlingHandler);
70+
71+
channel.writeInbound("1", "2", "3", "4", "5");
72+
73+
// the downstream handler disabled auto-read while processing the first message; the remaining messages
74+
// must be held in the flow control handler's queue instead of being delivered
75+
assertEquals(throttlingHandler.received.size(), 1);
76+
assertFalse(flowControlHandler.isQueueEmpty());
77+
78+
// re-enabling auto-read resumes delivery; the downstream handler pauses again on the next message,
79+
// so exactly one more message must be delivered (per-message auto-read granularity)
80+
throttlingHandler.pauseAgainAtCount(2);
81+
channel.config().setAutoRead(true);
82+
assertEquals(throttlingHandler.received.size(), 2);
83+
assertFalse(flowControlHandler.isQueueEmpty());
84+
85+
// with auto-read left enabled, the rest of the queue drains
86+
throttlingHandler.pauseAgainAtCount(-1);
87+
channel.config().setAutoRead(true);
88+
assertEquals(throttlingHandler.received.size(), 5);
89+
assertTrue(flowControlHandler.isQueueEmpty());
90+
91+
assertFalse(channel.finish());
92+
}
93+
94+
@Test
95+
public void shouldDeliverOneMessagePerReadWhenAutoReadIsDisabled() {
96+
PulsarFlowControlHandler flowControlHandler = new PulsarFlowControlHandler();
97+
ThrottlingHandler throttlingHandler = new ThrottlingHandler(0);
98+
EmbeddedChannel channel = new EmbeddedChannel(flowControlHandler, throttlingHandler);
99+
channel.config().setAutoRead(false);
100+
101+
// channel activation (with auto-read initially enabled) issued one read() through the pipeline, so there
102+
// is one outstanding message of read demand that the first written message satisfies
103+
channel.writeInbound("1", "2", "3");
104+
assertEquals(throttlingHandler.received.size(), 1);
105+
106+
channel.read();
107+
assertEquals(throttlingHandler.received.size(), 2);
108+
109+
channel.read();
110+
assertEquals(throttlingHandler.received.size(), 3);
111+
assertTrue(flowControlHandler.isQueueEmpty());
112+
113+
assertFalse(channel.finish());
114+
}
115+
}

0 commit comments

Comments
 (0)