PIP-36: Support set message size in broker.conf #4247

zymap · 2019-05-09T12:55:07Z

Motivation

Currently Pulsar only support 5MB size of messages.But there are many cases will use more than 5MB message
to transfer.
https://github.com/apache/pulsar/wiki/PIP-36%3A-Max-Message-Size

Modifications

Add message size in protocol
Automaticlly adjust client message size by server

--- *Motivation* Currently Pulsar only support 5MB size of messages.But there are many cases will use more than 5MB message to transfer. https://github.com/apache/pulsar/wiki/PIP-36%3A-Max-Message-Size *Modifications* - Add message size in protocol - Automaticlly adjust client message size by server

zymap · 2019-05-09T12:55:52Z

I will add test later. @sijie PTAL.

merlimat · 2019-05-09T16:59:14Z

pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java

@@ -139,6 +140,15 @@ private static boolean argsContains(String[] args, String arg) {
                brokerConfig = loadConfig(starterArguments.brokerConfigFile);
            }

+            int maxFrameSize = brokerConfig.getMaxMessageSize() + (10 * 1024);
+            if (maxFrameSize < 0) {


The annotations in ServiceConfiguration already have a way to enforce values limits

Oh, will replace it

merlimat · 2019-05-09T17:00:27Z

pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java

+            if (maxFrameSize < 0) {
+                throw new IllegalArgumentException("Max message size need smaller than 5233640 bytes");
+            }
+            if (maxFrameSize > DirectMemoryUtils.maxDirectMemory()) {


This check is not going to be very useful. Even if we have 1 single message of a size comparable to maxDirectMem, we'd be already in trouble

Ok, will fix it as >=

merlimat · 2019-05-09T17:01:53Z

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

+    @FieldContext(category = CATEGORY_SERVER, doc = "The size of messages.")
+    private int maxMessageSize = 5 * 1024 * 1024;
+
+    private int maxFrameSize = 5242880;


Having 2 settings, which are ultimately strongly related, could be confusing to users. I think the max message size with some padding for headers would be easier in general.

merlimat · 2019-05-09T17:06:54Z

pulsar-common/src/main/proto/PulsarApi.proto

@@ -213,6 +213,7 @@ message CommandConnect {
 message CommandConnected {
 	required string server_version = 1;
 	optional int32 protocol_version = 2 [default = 0];
+	optional int32 max_message_size = 3 [default = 5242880];


For this kind of numbers, I'd leave the default empty and let the client/broker assume the 5MB default.

In particular, if the default is in Java code, we can put in a constant shared between client and broker and keep it in a single place

Yeah, you are right. will fix it

merlimat · 2019-05-09T17:07:42Z

...iscovery-service/src/main/java/org/apache/pulsar/discovery/service/server/ServiceConfig.java

+        this.maxMessageSize = maxMessageSize;
+    }
+
+    private int maxMessageSize = 5242880;


This should be defined in a static constant, shared between client and broker.

merlimat · 2019-05-09T17:08:15Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java

@@ -99,7 +99,8 @@ protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc()));
                }
                ch.pipeline().addLast("frameDecoder",
-                        new LengthFieldBasedFrameDecoder(PulsarDecoder.MaxFrameSize, 0, 4, 0, 4));
+                                      new LengthFieldBasedFrameDecoder(config.getMaxMessagesSize() + 10 * 1024, 0, 4, 0,


The amount of padding should also be defined in a constant (shared between all components)

such as pulsar-common?

Yes, somewhere like Commands.MESSAGE_SIZE_FRAME_PADDING

merlimat · 2019-05-09T17:09:25Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java

@@ -218,7 +218,8 @@ private void completeConnect() {
            // partitions metadata lookups
            state = State.ProxyLookupRequests;
            lookupProxyHandler = new LookupProxyHandler(service, this);
-            ctx.writeAndFlush(Commands.newConnected(protocolVersionToAdvertise));
+            ctx.writeAndFlush(
+                Commands.newConnected(protocolVersionToAdvertise, service.getConfiguration().getMaxMessagesSize()));


Proxy should also take into consideration what's the min between proxy and broker.

Eg: if proxy has 5MB and broker has 1MB, proxy should downgrade itself to 1MB

Oh, right, proxy is a client of broker so it need receive the max message size from broker and adjust itself. Will fix

Yes, the proxy should not even need the config option. Just take the broker value.

Yeah, already fixed in the last commit

merlimat · 2019-05-09T17:09:41Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ServiceChannelInitializer.java

@@ -85,7 +85,8 @@ protected void initChannel(SocketChannel ch) throws Exception {
            }
        }

-        ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(PulsarDecoder.MaxFrameSize, 0, 4, 0, 4));
+        ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(
+            proxyService.getConfiguration().getMaxMessagesSize() + 10 * 1024, 0, 4, 0, 4));


padding from constant

merlimat · 2019-05-09T17:57:50Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java

@@ -62,6 +62,7 @@
    private int connectionTimeoutMs = 10000;
    private long defaultBackoffIntervalNanos = TimeUnit.MILLISECONDS.toNanos(100);
    private long maxBackoffIntervalNanos = TimeUnit.SECONDS.toNanos(30);
+    private int defaultMaxFrameSize = 5242880;


The default should come from constant (and not be modifiable)

Actually, this is not needed anymore. The client will not need to configure this anyway

But it need initial value to init defaultFrameDecoder at first time and it will replace defaultFrameDecoder when receive command Connected immediately.

Oh I can use the value in Commands

merlimat · 2019-05-09T17:58:57Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

@@ -312,15 +312,15 @@ public void sendAsync(Message<T> message, SendCallback callback) {

            // validate msg-size (For batching this will be check at the batch completion size)
            int compressedSize = compressedPayload.readableBytes();
-
-            if (compressedSize > PulsarDecoder.MaxMessageSize) {
+            int maxMessageSize = ClientCnx.NUMBER_OF_MAX_MESSAGE_SIZE.get(this.cnx());


This should be exposed in the ClientCnx directly, like: cnx.getMaxMessageSize()

merlimat · 2019-05-09T18:00:22Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

@@ -275,7 +280,10 @@ protected void handleConnected(CommandConnected connected) {
        }

        checkArgument(state == State.SentConnectFrame || state == State.Connecting);
-
+        int maxFrameSize = connected.getMaxMessageSize() - 10 * 1024;
+        NUMBER_OF_MAX_MESSAGE_SIZE.compareAndSet(this, 0, maxFrameSize);


This compare and set is not needed here. The CommandConnected is only received once, so we should just do the replace of the frame decoder.

merlimat · 2019-05-09T18:01:08Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

@@ -275,7 +280,10 @@ protected void handleConnected(CommandConnected connected) {
        }

        checkArgument(state == State.SentConnectFrame || state == State.Connecting);
-
+        int maxFrameSize = connected.getMaxMessageSize() - 10 * 1024;


Padding from constant (and also it should be + rather than -

merlimat · 2019-05-09T18:01:54Z

Broker should also set the max entry size on the BK client instance

zymap · 2019-05-10T03:10:42Z

@merlimat Thank you for your review. I will fix those problem as you said. :)

--- *Motivation* When broker specify a `maxMessageSize` bookie should accept this value as `nettyFrameSize` *Modifications* - Use `cnx().getMaxMessageSize` - Discovery service only redirect so use the constant value `5 * 1024 * 1024` as message size - Put `MAX_METADATA_SIZE` as constant value in `InternalConfigurationData`

zymap · 2019-05-10T11:07:00Z

@merlimat PTAL.

merlimat · 2019-05-10T17:46:06Z

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

+        category = CATEGORY_SERVER,
+        doc = "Max size of messages.",
+        maxValue = Integer.MAX_VALUE - InternalConfigurationData.MESSAGE_META_SIZE)
+    private int maxMessageSize = 5 * 1024 * 1024;


Use constant for default max size. eg: Commands.DEFAULT_MAX_MESSAGE_SIZE

merlimat · 2019-05-10T17:48:11Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

@@ -275,7 +282,9 @@ protected void handleConnected(CommandConnected connected) {
        }

        checkArgument(state == State.SentConnectFrame || state == State.Connecting);
-
+        this.maxMessageSize = connected.getMaxMessageSize();


if (connected.hasMaxMessageSize() { . .. } else { maxMessageSize = Commands.DEFAULT_MAX_MESSAGE_SIZE; }

Thank you for reminder. It' s the optional value in protocol, will fix

merlimat · 2019-05-10T17:49:15Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java

@@ -62,6 +62,7 @@
    private int connectionTimeoutMs = 10000;
    private long defaultBackoffIntervalNanos = TimeUnit.MILLISECONDS.toNanos(100);
    private long maxBackoffIntervalNanos = TimeUnit.SECONDS.toNanos(30);
+    private int defaultMaxFrameSize = 5242880;


Actually, this is not needed anymore. The client will not need to configure this anyway

merlimat · 2019-05-10T17:51:42Z

pulsar-common/src/main/java/org/apache/pulsar/common/conf/InternalConfigurationData.java

@@ -23,6 +23,7 @@

 public class InternalConfigurationData {

+    public final static int MESSAGE_META_SIZE = 10 * 1024;


This class is a POJO used to serialize to JSON.

I'd keep this constant in Commands where all the binary protocol stuff is.
eg: Commands.MESSAGE_SIZE_FRAME_PADDING

along with Commands.DEFAULT_MAX_MESSAGE_SIZE

Ok, will fix

merlimat · 2019-05-10T17:52:34Z

...ar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/ServerConnection.java

@@ -101,7 +102,7 @@ protected void handleConnect(CommandConnect connect) {
                return;
            }
        }
-        ctx.writeAndFlush(Commands.newConnected(connect.getProtocolVersion()));
+        ctx.writeAndFlush(Commands.newConnected(connect.getProtocolVersion(), ServiceConfig.MAX_MESSAGE_SIZE));


Commands.DEFAULT_MAX_MESSAGE_SIZE

I think discovery service only for looking up service so it's doesn't need set max message size, right?

merlimat · 2019-05-10T17:53:26Z

...iscovery-service/src/main/java/org/apache/pulsar/discovery/service/server/ServiceConfig.java

@@ -34,6 +33,10 @@
 */
 public class ServiceConfig implements PulsarConfiguration {

+    // Discovery service doesn't send any messages except Command connected.
+    // So it's ok use a default value.
+    public final static int MAX_MESSAGE_SIZE = 5 * 1024 * 1024;


Don't need to redefine the same constant. Just reuse Commands.DEFAULT_MAX_MESSAGE_SIZE

merlimat · 2019-05-10T17:54:30Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java

@@ -111,6 +111,9 @@
    )
    private String brokerWebServiceURLTLS;

+    @FieldContext(category = CATEGORY_BROKER_DISCOVERY, doc = "")
+    private int maxMessagesSize = 5233640;


Proxy should not have this config option. Just repeat what broker had.

Yeah. will fix

--- *Modifications* - use `Commands` to store default `MAX_MESSAGE_SIZE` and `MESSAGE_SIZE_FRAME_PADDING` - replace `LengthFieldBasedFrameDecoder` when has set message size - replace `PulsarDecoder.MaxMessageSize`

zymap · 2019-05-13T09:02:45Z

@merlimat I do some changes. Could you please take a look?

merlimat

LGTM. Just a couple of minor comments

merlimat · 2019-05-15T01:22:10Z

pulsar-common/src/main/java/org/apache/pulsar/common/conf/InternalConfigurationData.java

@@ -23,6 +23,7 @@

 public class InternalConfigurationData {

+    public final static int MESSAGE_META_SIZE = 10 * 1024;


This can be removed now.

merlimat · 2019-05-15T01:23:47Z

pulsar-common/src/main/java/org/apache/pulsar/common/api/Commands.java

@@ -189,9 +194,16 @@ public static ByteBuf newConnect(String authMethodName, AuthData authData, int p
        return res;
    }

-    public static ByteBuf newConnected(int clientProtocolVersion) {
+    public static ByteBuf newConnected(int clientProtocoVersion) {
+        return newConnected(clientProtocoVersion, INVALID_MAX_MESSAGE_SIZE);


Why do we need the INVALID_MAX_MESSAGE_SIZE? Is this method called from unit tests? I'd rather just leave the method which requires the maxMessageSize

Because it will be used when there no max message size in protocol, like this:

ChannelFuture channelFuture; if (connected.hasMaxMessageSize()) { channelFuture = inboundChannel.writeAndFlush( Commands.newConnected(connected.getProtocolVersion(), connected.getMaxMessageSize())); } else { channelFuture = inboundChannel.writeAndFlush(Commands.newConnected(connected.getProtocolVersion())); }

merlimat · 2019-05-15T01:25:44Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java

+                                                                                    ParserProxyHandler.BACKEND_CONN,
+                                                                                    connected.getMaxMessageSize()));
+                    } else {
+                    inboundChannel.pipeline().addBefore("handler", "inboundParser",


nit: the block should be indented

--- *Motivation* - Even if the cnx can't use, `maxMessageSize` should be used at compare message size. So it should as a static variable

merlimat

👍

merlimat · 2019-05-15T19:16:43Z

run java8 tests

Update the Reference > Pulsar configuration > broker for Pulsar 2.4.0 release apache#4247

This update is for the PR: #4247 Doc update for the master is approved and merged. Update the Reference > Pulsar configuration > broker for the following releases: 2.5.2, 2.5.1, 2.5.0, 2.4.2, 2.4.1, and 2.4.0.

This update is for the PR: apache#4247 Doc update for the master is approved and merged. Update the Reference > Pulsar configuration > broker for the following releases: 2.5.2, 2.5.1, 2.5.0, 2.4.2, 2.4.1, and 2.4.0.

zymap changed the title ~~Support set message size in broker.conf~~ [WIP] Support set message size in broker.conf May 9, 2019

zymap changed the title ~~[WIP] Support set message size in broker.conf~~ [WIP] PIP-36: Support set message size in broker.conf May 9, 2019

merlimat reviewed May 9, 2019

View reviewed changes

merlimat reviewed May 10, 2019

View reviewed changes

zymap added 2 commits May 13, 2019 16:49

Use Commands to store message setting

53f8ae2

--- *Modifications* - use `Commands` to store default `MAX_MESSAGE_SIZE` and `MESSAGE_SIZE_FRAME_PADDING` - replace `LengthFieldBasedFrameDecoder` when has set message size - replace `PulsarDecoder.MaxMessageSize`

Fix some error

99a2b44

Fix license header

59f8d3b

merlimat reviewed May 15, 2019

View reviewed changes

zymap added 3 commits May 15, 2019 16:57

Add test and make ClientCnx.maxMessageSize static

70edbb3

--- *Motivation* - Even if the cnx can't use, `maxMessageSize` should be used at compare message size. So it should as a static variable

fix code style

8847c32

Fix license header

fb83d3f

merlimat added this to the 2.4.0 milestone May 15, 2019

merlimat added the type/enhancement The enhancements for the existing features or docs. e.g. reduce memory usage of the delayed messages label May 15, 2019

merlimat approved these changes May 15, 2019

View reviewed changes

merlimat merged commit da1cd78 into apache:master May 15, 2019

sijie assigned zymap May 19, 2019

sijie changed the title ~~[WIP] PIP-36: Support set message size in broker.conf~~ PIP-36: Support set message size in broker.conf May 19, 2019

sijie mentioned this pull request May 19, 2019

Message size is bigger than 5232640 bytes #4243

Closed

sijie added area/broker area/client area/proxy labels May 19, 2019

sijie added the component/storage label May 19, 2019

zymap mentioned this pull request May 23, 2019

Exposing MaxMessageSize to user configuration #4054

Closed

jiazhai mentioned this pull request Jun 19, 2019

PIP 37: [pulsar-client] support large message size #4400

Merged

Huanli-Meng added a commit to Huanli-Meng/pulsar that referenced this pull request Jun 4, 2020

[Doc] Update for the maximum message size

0f53f7f

Update the Reference > Pulsar configuration > broker for Pulsar 2.4.0 release apache#4247

Huanli-Meng mentioned this pull request Jun 4, 2020

[Doc] Update doc for the maximum message size #7171

Merged

1 task

		@@ -23,6 +23,7 @@

		public class InternalConfigurationData {

		public final static int MESSAGE_META_SIZE = 10 * 1024;

PIP-36: Support set message size in broker.conf #4247

PIP-36: Support set message size in broker.conf #4247

Conversation

zymap commented May 9, 2019

zymap commented May 9, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

merlimat commented May 9, 2019

zymap commented May 10, 2019

zymap commented May 10, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zymap commented May 13, 2019

merlimat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

merlimat left a comment

Choose a reason for hiding this comment

merlimat commented May 15, 2019