You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/08/14 15:40:35 UTC
(pulsar) 04/04: [fix][proxy] Fix TooLongFrameException with Pulsar Proxy (#24626)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ba3b6cb522299d16fedc15dc8e7aa8e658794e2d
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Aug 14 17:11:45 2025 +0300
[fix][proxy] Fix TooLongFrameException with Pulsar Proxy (#24626)
(cherry picked from commit cde4948044767033e40bc7d4eff98a4c25fd3aff)
---
.../broker/service/PulsarChannelInitializer.java | 6 +-
.../pulsar/client/api/MockBrokerService.java | 4 +-
.../org/apache/pulsar/client/impl/ClientCnx.java | 5 +-
.../client/impl/PulsarChannelInitializer.java | 8 +--
.../pulsar/common/protocol/FrameDecoderUtil.java | 67 ++++++++++++++++++++++
.../pulsar/proxy/server/DirectProxyHandler.java | 25 +++-----
.../proxy/server/ServiceChannelInitializer.java | 7 +--
7 files changed, 86 insertions(+), 36 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java
index f15f6d67766..7a8d05b906a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java
@@ -22,7 +22,6 @@ import com.google.common.annotations.VisibleForTesting;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
-import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.flow.FlowControlHandler;
import io.netty.handler.flush.FlushConsolidationHandler;
import io.netty.handler.ssl.SslContext;
@@ -34,7 +33,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.common.protocol.ByteBufPair;
-import org.apache.pulsar.common.protocol.Commands;
+import org.apache.pulsar.common.protocol.FrameDecoderUtil;
import org.apache.pulsar.common.protocol.OptionalProxyProtocolDecoder;
import org.apache.pulsar.common.util.NettyServerSslContextBuilder;
import org.apache.pulsar.common.util.SslContextAutoRefreshBuilder;
@@ -122,8 +121,7 @@ public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel>
if (pulsar.getConfiguration().isHaProxyProtocolEnabled()) {
ch.pipeline().addLast(OptionalProxyProtocolDecoder.NAME, new OptionalProxyProtocolDecoder());
}
- ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(
- brokerConf.getMaxMessageSize() + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0, 4));
+ FrameDecoderUtil.addFrameDecoder(ch.pipeline(), brokerConf.getMaxMessageSize());
// https://stackoverflow.com/questions/37535482/netty-disabling-auto-read-doesnt-work-for-bytetomessagedecoder
// Classes such as {@link ByteToMessageDecoder} or {@link MessageToByteEncoder} are free to emit as many events
// as they like for any given input. so, disabling auto-read on `ByteToMessageDecoder` doesn't work properly and
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java
index 9ca0bafe00b..cb55e9c48b9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerService.java
@@ -28,7 +28,6 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.SocketChannel;
-import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import java.io.IOException;
import java.net.InetSocketAddress;
@@ -69,6 +68,7 @@ import org.apache.pulsar.common.api.proto.CommandUnsubscribe;
import org.apache.pulsar.common.lookup.data.LookupData;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.protocol.Commands;
+import org.apache.pulsar.common.protocol.FrameDecoderUtil;
import org.apache.pulsar.common.protocol.PulsarDecoder;
import org.apache.pulsar.common.protocol.schema.SchemaVersion;
import org.apache.pulsar.common.util.netty.EventLoopUtil;
@@ -338,7 +338,7 @@ public class MockBrokerService {
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
- ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(MaxMessageSize, 0, 4, 0, 4));
+ FrameDecoderUtil.addFrameDecoder(ch.pipeline(), MaxMessageSize);
ch.pipeline().addLast("handler", new MockServerCnx());
}
});
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
index 6d157802f71..39417a8dd0d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
@@ -29,7 +29,6 @@ import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.unix.Errors.NativeIoException;
-import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.ssl.SslHandshakeCompletionEvent;
import io.netty.util.concurrent.Promise;
import java.net.InetSocketAddress;
@@ -97,6 +96,7 @@ import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate;
import org.apache.pulsar.common.api.proto.ServerError;
import org.apache.pulsar.common.lookup.GetTopicsResult;
import org.apache.pulsar.common.protocol.Commands;
+import org.apache.pulsar.common.protocol.FrameDecoderUtil;
import org.apache.pulsar.common.protocol.PulsarHandler;
import org.apache.pulsar.common.protocol.schema.SchemaVersion;
import org.apache.pulsar.common.schema.SchemaInfo;
@@ -379,8 +379,7 @@ public class ClientCnx extends PulsarHandler {
+ "server frame size {}", ctx.channel(), connected.getMaxMessageSize());
}
maxMessageSize = connected.getMaxMessageSize();
- ctx.pipeline().replace("frameDecoder", "newFrameDecoder", new LengthFieldBasedFrameDecoder(
- connected.getMaxMessageSize() + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0, 4));
+ FrameDecoderUtil.replaceFrameDecoder(ctx.pipeline(), connected.getMaxMessageSize());
}
if (log.isDebugEnabled()) {
log.debug("{} Connection is ready", ctx.channel());
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java
index 6385ab78ccd..31e559ebfb6 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java
@@ -22,7 +22,6 @@ import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
-import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.flush.FlushConsolidationHandler;
import io.netty.handler.proxy.Socks5ProxyHandler;
import io.netty.handler.ssl.SslContext;
@@ -44,6 +43,7 @@ import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.client.util.ObjectCache;
import org.apache.pulsar.common.protocol.ByteBufPair;
import org.apache.pulsar.common.protocol.Commands;
+import org.apache.pulsar.common.protocol.FrameDecoderUtil;
import org.apache.pulsar.common.util.SecurityUtility;
import org.apache.pulsar.common.util.keystoretls.NettySSLContextAutoRefreshBuilder;
import org.apache.pulsar.common.util.netty.NettyFutureUtil;
@@ -89,9 +89,7 @@ public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel>
// Setup channel except for the SsHandler for TLS enabled connections
ch.pipeline().addLast("ByteBufPairEncoder", ByteBufPair.getEncoder(tlsEnabled));
-
- ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(
- Commands.DEFAULT_MAX_MESSAGE_SIZE + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0, 4));
+ FrameDecoderUtil.addFrameDecoder(ch.pipeline(), Commands.DEFAULT_MAX_MESSAGE_SIZE);
ChannelHandler clientCnx = clientCnxSupplier.get();
ch.pipeline().addLast("handler", clientCnx);
}
@@ -165,7 +163,7 @@ public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel>
}, TLS_CERTIFICATE_CACHE_MILLIS, TimeUnit.MILLISECONDS));
}
- /**
+ /**
* Initialize TLS for a channel. Should be invoked before the channel is connected to the remote address.
*
* @param ch the channel
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/FrameDecoderUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/FrameDecoderUtil.java
new file mode 100644
index 00000000000..aee9f2c39bd
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/FrameDecoderUtil.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.protocol;
+
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
+import lombok.experimental.UtilityClass;
+
+/**
+ * Utility class for managing Netty LenghtFieldBasedFrameDecoder instances in a Netty ChannelPipeline
+ * for the Pulsar binary protocol.
+ */
+@UtilityClass
+public class FrameDecoderUtil {
+ public static final String FRAME_DECODER_HANDLER = "frameDecoder";
+
+ /**
+ * Adds a LengthFieldBasedFrameDecoder to the given ChannelPipeline.
+ *
+ * @param pipeline the ChannelPipeline to which the decoder will be added
+ * @param maxMessageSize the maximum size of messages that can be decoded
+ */
+ public static void addFrameDecoder(ChannelPipeline pipeline, int maxMessageSize) {
+ pipeline.addLast(FRAME_DECODER_HANDLER, createFrameDecoder(maxMessageSize));
+ }
+
+ /**
+ * Replaces the existing LengthFieldBasedFrameDecoder in the given ChannelPipeline with a new one.
+ *
+ * @param pipeline the ChannelPipeline in which the decoder will be replaced
+ * @param maxMessageSize the maximum size of messages that can be decoded
+ */
+ public static void replaceFrameDecoder(ChannelPipeline pipeline, int maxMessageSize) {
+ pipeline.replace(FRAME_DECODER_HANDLER, FRAME_DECODER_HANDLER, createFrameDecoder(maxMessageSize));
+ }
+
+ /**
+ * Removes the LengthFieldBasedFrameDecoder from the given ChannelPipeline.
+ * This is useful in the Pulsar Proxy to remove the decoder before direct proxying of messages without decoding.
+ *
+ * @param pipeline the ChannelPipeline from which the decoder will be removed
+ */
+ public static void removeFrameDecoder(ChannelPipeline pipeline) {
+ pipeline.remove(FRAME_DECODER_HANDLER);
+ }
+
+ private static LengthFieldBasedFrameDecoder createFrameDecoder(int maxMessageSize) {
+ return new LengthFieldBasedFrameDecoder(
+ maxMessageSize + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0, 4);
+ }
+}
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java
index 7c1d33a78e4..ac5af17f273 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java
@@ -34,7 +34,6 @@ import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.epoll.EpollMode;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.socket.SocketChannel;
-import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.haproxy.HAProxyCommand;
import io.netty.handler.codec.haproxy.HAProxyMessage;
import io.netty.handler.codec.haproxy.HAProxyProtocolVersion;
@@ -61,6 +60,7 @@ import org.apache.pulsar.common.api.proto.CommandAuthChallenge;
import org.apache.pulsar.common.api.proto.CommandConnected;
import org.apache.pulsar.common.api.proto.FeatureFlags;
import org.apache.pulsar.common.protocol.Commands;
+import org.apache.pulsar.common.protocol.FrameDecoderUtil;
import org.apache.pulsar.common.protocol.PulsarDecoder;
import org.apache.pulsar.common.stats.Rate;
import org.apache.pulsar.common.util.NettyClientSslContextRefresher;
@@ -205,9 +205,7 @@ public class DirectProxyHandler {
ch.pipeline().addLast("readTimeoutHandler",
new ReadTimeoutHandler(brokerProxyReadTimeoutMs, TimeUnit.MILLISECONDS));
}
- ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(
- service.getConfiguration().getMaxMessageSize() + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0,
- 4));
+ FrameDecoderUtil.addFrameDecoder(ch.pipeline(), service.getConfiguration().getMaxMessageSize());
ch.pipeline().addLast("proxyOutboundHandler",
(ChannelHandler) new ProxyBackendHandler(config, protocolVersion, remoteHost, featureFlags));
}
@@ -447,23 +445,16 @@ public class DirectProxyHandler {
log.debug("[{}] [{}] Removing decoder from pipeline", inboundChannel, outboundChannel);
}
// direct tcp proxy
- inboundChannel.pipeline().remove("frameDecoder");
- outboundChannel.pipeline().remove("frameDecoder");
+ FrameDecoderUtil.removeFrameDecoder(inboundChannel.pipeline());
+ FrameDecoderUtil.removeFrameDecoder(outboundChannel.pipeline());
} else {
// Enable parsing feature, proxyLogLevel(1 or 2)
// Add parser handler
if (connected.hasMaxMessageSize()) {
- inboundChannel.pipeline()
- .replace("frameDecoder", "newFrameDecoder",
- new LengthFieldBasedFrameDecoder(connected.getMaxMessageSize()
- + Commands.MESSAGE_SIZE_FRAME_PADDING,
- 0, 4, 0, 4));
- outboundChannel.pipeline().replace("frameDecoder", "newFrameDecoder",
- new LengthFieldBasedFrameDecoder(
- connected.getMaxMessageSize()
- + Commands.MESSAGE_SIZE_FRAME_PADDING,
- 0, 4, 0, 4));
-
+ FrameDecoderUtil.replaceFrameDecoder(inboundChannel.pipeline(),
+ connected.getMaxMessageSize());
+ FrameDecoderUtil.replaceFrameDecoder(outboundChannel.pipeline(),
+ connected.getMaxMessageSize());
inboundChannel.pipeline().addBefore("handler", "inboundParser",
new ParserProxyHandler(service,
ParserProxyHandler.FRONTEND_CONN,
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ServiceChannelInitializer.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ServiceChannelInitializer.java
index 19f4002ad52..acae088f1c7 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ServiceChannelInitializer.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ServiceChannelInitializer.java
@@ -20,14 +20,13 @@ package org.apache.pulsar.proxy.server;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
-import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.flush.FlushConsolidationHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.ssl.SslProvider;
import io.netty.handler.timeout.ReadTimeoutHandler;
import java.util.concurrent.TimeUnit;
-import org.apache.pulsar.common.protocol.Commands;
+import org.apache.pulsar.common.protocol.FrameDecoderUtil;
import org.apache.pulsar.common.protocol.OptionalProxyProtocolDecoder;
import org.apache.pulsar.common.util.NettyServerSslContextBuilder;
import org.apache.pulsar.common.util.SslContextAutoRefreshBuilder;
@@ -112,9 +111,7 @@ public class ServiceChannelInitializer extends ChannelInitializer<SocketChannel>
if (proxyService.getConfiguration().isHaProxyProtocolEnabled()) {
ch.pipeline().addLast(OptionalProxyProtocolDecoder.NAME, new OptionalProxyProtocolDecoder());
}
- ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(
- this.maxMessageSize + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0, 4));
-
+ FrameDecoderUtil.addFrameDecoder(ch.pipeline(), maxMessageSize);
ch.pipeline().addLast("handler", new ProxyConnection(proxyService, proxyService.getDnsAddressResolverGroup()));
}
}