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

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

@@ -33,6 +33,8 @@
 import lombok.Setter;
 import org.apache.bookkeeper.client.api.DigestType;
 import org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider;
+import org.apache.pulsar.common.api.Commands;
+import org.apache.pulsar.common.conf.InternalConfigurationData;
 import org.apache.pulsar.common.configuration.Category;
 import org.apache.pulsar.common.configuration.FieldContext;
 import org.apache.pulsar.common.configuration.PulsarConfiguration;
@@ -489,6 +491,12 @@ public class ServiceConfiguration implements PulsarConfiguration {
             + " Using a value of 0, is disabling maxConsumersPerSubscription-limit check.")
     private int maxConsumersPerSubscription = 0;
 
+    @FieldContext(
+        category = CATEGORY_SERVER,
+        doc = "Max size of messages.",
+        maxValue = Integer.MAX_VALUE - Commands.MESSAGE_SIZE_FRAME_PADDING)
+    private int maxMessageSize = Commands.DEFAULT_MAX_MESSAGE_SIZE;
+
     /***** --- TLS --- ****/
     @FieldContext(
         category = CATEGORY_TLS,

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

@@ -45,10 +45,13 @@
 import org.apache.bookkeeper.replication.AutoRecoveryMain;
 import org.apache.bookkeeper.stats.StatsProvider;
 import org.apache.bookkeeper.common.util.ReflectionUtils;
+import org.apache.bookkeeper.util.DirectMemoryUtils;
 import org.apache.commons.configuration.ConfigurationException;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.ServiceConfigurationUtils;
+import org.apache.pulsar.common.api.Commands;
+import org.apache.pulsar.common.conf.InternalConfigurationData;
 import org.apache.pulsar.functions.worker.WorkerConfig;
 import org.apache.pulsar.functions.worker.WorkerService;
 import org.slf4j.Logger;
@@ -139,6 +142,11 @@ private static class BrokerStarter {
                 brokerConfig = loadConfig(starterArguments.brokerConfigFile);
             }
 
+            int maxFrameSize = brokerConfig.getMaxMessageSize() + Commands.MESSAGE_SIZE_FRAME_PADDING;
+            if (maxFrameSize >= DirectMemoryUtils.maxDirectMemory()) {
+                throw new IllegalArgumentException("Max message size need smaller than jvm directMemory");
+            }
+
             // init functions worker
             if (starterArguments.runFunctionsWorker || brokerConfig.isFunctionsWorkerEnabled()) {
                 WorkerConfig workerConfig;

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

@@ -28,6 +28,8 @@
 import org.apache.bookkeeper.client.RackawareEnsemblePlacementPolicy;
 import org.apache.bookkeeper.client.RegionAwareEnsemblePlacementPolicy;
 import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.pulsar.common.api.Commands;
+import org.apache.pulsar.common.conf.InternalConfigurationData;
 import org.apache.pulsar.zookeeper.ZkBookieRackAffinityMapping;
 import org.apache.pulsar.zookeeper.ZkIsolatedBookieEnsemblePlacementPolicy;
 import org.apache.pulsar.zookeeper.ZooKeeperCache;
@@ -57,6 +59,7 @@ public BookKeeper create(ServiceConfiguration conf, ZooKeeper zkClient) throws I
         bkConf.setUseV2WireProtocol(conf.isBookkeeperUseV2WireProtocol());
         bkConf.setEnableDigestTypeAutodetection(true);
         bkConf.setStickyReadsEnabled(conf.isBookkeeperEnableStickyReads());
+        bkConf.setNettyMaxFrameSizeBytes(conf.getMaxMessageSize() + Commands.MESSAGE_SIZE_FRAME_PADDING);
 
         bkConf.setAllocatorPoolingPolicy(PoolingPolicy.UnpooledHeap);
         if (conf.isBookkeeperClientHealthCheckEnabled()) {

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

@@ -21,7 +21,8 @@
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.common.api.ByteBufPair;
-import org.apache.pulsar.common.api.PulsarDecoder;
+import org.apache.pulsar.common.api.Commands;
+import org.apache.pulsar.common.conf.InternalConfigurationData;
 import org.apache.pulsar.common.util.NettySslContextBuilder;
 
 import io.netty.channel.ChannelInitializer;
@@ -35,6 +36,7 @@ public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel>
     private final PulsarService pulsar;
     private final boolean enableTls;
     private final NettySslContextBuilder sslCtxRefresher;
+    private final ServiceConfiguration brokerConf;
 
     /**
      *
@@ -54,6 +56,7 @@ public PulsarChannelInitializer(PulsarService pulsar, boolean enableTLS) throws
         } else {
             this.sslCtxRefresher = null;
         }
+        this.brokerConf = pulsar.getConfiguration();
     }
 
     @Override
@@ -65,7 +68,8 @@ protected void initChannel(SocketChannel ch) throws Exception {
             ch.pipeline().addLast("ByteBufPairEncoder", ByteBufPair.ENCODER);
         }
 
-        ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(PulsarDecoder.MaxFrameSize, 0, 4, 0, 4));
+        ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(
+            brokerConf.getMaxMessageSize() + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0, 4));
         ch.pipeline().addLast("handler", new ServerCnx(pulsar));
     }
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -136,6 +136,7 @@ public class ServerCnx extends PulsarHandler {
     private boolean authenticateOriginalAuthData;
     private final boolean schemaValidationEnforced;
     private String authMethod = "none";
+    private final int maxMessageSize;
 
     enum State {
         Start, Connected, Failed, Connecting
@@ -156,6 +157,7 @@ public ServerCnx(PulsarService pulsar) {
         this.proxyRoles = service.pulsar().getConfiguration().getProxyRoles();
         this.authenticateOriginalAuthData = service.pulsar().getConfiguration().isAuthenticateOriginalAuthData();
         this.schemaValidationEnforced = pulsar.getConfiguration().isSchemaValidationEnforced();
+        this.maxMessageSize = pulsar.getConfiguration().getMaxMessageSize();
     }
 
     @Override
@@ -455,7 +457,7 @@ private String getOriginalPrincipal(String originalAuthData, String originalAuth
 
     // complete the connect and sent newConnected command
     private void completeConnect(int clientProtoVersion, String clientVersion) {
-        ctx.writeAndFlush(Commands.newConnected(clientProtoVersion));
+        ctx.writeAndFlush(Commands.newConnected(clientProtoVersion, maxMessageSize));
         state = State.Connected;
         remoteEndpointProtocolVersion = clientProtoVersion;
         if (isNotBlank(clientVersion) && !clientVersion.contains(" ") /* ignore default version: pulsar client */) {

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MaxMessageSizeTest.java

@@ -0,0 +1,154 @@
+/**
+ * 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.broker.service;
+
+import com.google.common.collect.Sets;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.test.PortManager;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class MaxMessageSizeTest {
+    private static int BROKER_SERVICE_PORT = PortManager.nextFreePort();
+    PulsarService pulsar;
+    ServiceConfiguration configuration;
+
+    PulsarAdmin admin;
+
+    LocalBookkeeperEnsemble bkEnsemble;
+
+    private final int ZOOKEEPER_PORT = PortManager.nextFreePort();
+    private final int BROKER_WEBSERVER_PORT = PortManager.nextFreePort();
+
+    @BeforeMethod
+    void setup() {
+        try {
+            bkEnsemble = new LocalBookkeeperEnsemble(3, ZOOKEEPER_PORT, PortManager::nextFreePort);
+            ServerConfiguration conf = new ServerConfiguration();
+            conf.setNettyMaxFrameSizeBytes(10 * 1024 * 1024);
+            bkEnsemble.startStandalone(conf, false);
+
+            configuration = new ServiceConfiguration();
+            configuration.setZookeeperServers("127.0.0.1:" + ZOOKEEPER_PORT);
+            configuration.setAdvertisedAddress("localhost");
+            configuration.setWebServicePort(BROKER_WEBSERVER_PORT);
+            configuration.setClusterName("max_message_test");
+            configuration.setBrokerServicePort(BROKER_SERVICE_PORT);
+            configuration.setAuthorizationEnabled(false);
+            configuration.setAuthenticationEnabled(false);
+            configuration.setManagedLedgerMaxEntriesPerLedger(5);
+            configuration.setManagedLedgerMinLedgerRolloverTimeMinutes(0);
+            configuration.setMaxMessageSize(10 * 1024 * 1024);
+
+            pulsar = new PulsarService(configuration);
+            pulsar.start();
+
+            String url = "http://127.0.0.1:" + BROKER_WEBSERVER_PORT;
+            admin = PulsarAdmin.builder().serviceHttpUrl(url).build();
+            admin.clusters().createCluster("max_message_test", new ClusterData(url));
+            admin.tenants()
+                 .createTenant("test", new TenantInfo(Sets.newHashSet("appid1"), Sets.newHashSet("max_message_test")));
+            admin.namespaces().createNamespace("test/message", Sets.newHashSet("max_message_test"));
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    @AfterMethod
+    void shutdown() {
+        try {
+            pulsar.close();
+            bkEnsemble.stop();
+        } catch (Throwable t) {
+            t.printStackTrace();
+        }
+    }
+
+    @Test
+    public void testMaxMessageSetting() throws PulsarClientException {
+
+        PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://127.0.0.1:" + BROKER_SERVICE_PORT).build();
+        String topicName = "persistent://test/message/topic1";
+        Producer producer = client.newProducer().topic(topicName).sendTimeout(60, TimeUnit.SECONDS).create();
+        Consumer consumer = client.newConsumer().topic(topicName).subscriptionName("test1").subscribe();
+
+        // less than 5MB message
+
+        byte[] normalMsg = new byte[2 * 1024 * 1024];
+
+        try {
+            producer.send(normalMsg);
+        } catch (PulsarClientException e) {
+            Assert.fail("Shouldn't have exception at here", e);
+        }
+
+        byte[] consumerNormalMsg = consumer.receive().getData();
+        Assert.assertEquals(normalMsg, consumerNormalMsg);
+
+        // equal 5MB message
+        byte[] limitMsg = new byte[5 * 1024 * 1024];
+        try {
+            producer.send(limitMsg);
+        } catch (PulsarClientException e) {
+            Assert.fail("Shouldn't have exception at here", e);
+        }
+
+        byte[] consumerLimitMsg = consumer.receive().getData();
+        Assert.assertEquals(limitMsg, consumerLimitMsg);
+
+        // less than 10MB message
+        byte[] newNormalMsg = new byte[8 * 1024 * 1024];
+        try {
+            producer.send(newNormalMsg);
+        } catch (PulsarClientException e) {
+            Assert.fail("Shouldn't have exception at here", e);
+        }
+
+        byte[] consumerNewNormalMsg = consumer.receive().getData();
+        Assert.assertEquals(newNormalMsg, consumerNewNormalMsg);
+
+        // equals 10MB message
+        byte[] newLimitMsg = new byte[10 * 1024 * 1024];
+        try {
+            producer.send(newLimitMsg);
+            Assert.fail("Shouldn't send out this message");
+        } catch (PulsarClientException e) {
+            // no-op
+        }
+
+        consumer.unsubscribe();
+        consumer.close();
+        producer.close();
+        client.close();
+
+    }
+}

pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java

@@ -634,10 +634,10 @@ public void testSendBigMessageSize() throws Exception {
 
 
         // Messages are allowed up to MaxMessageSize
-        producer.newMessage().value(new byte[PulsarDecoder.MaxMessageSize]);
+        producer.newMessage().value(new byte[Commands.DEFAULT_MAX_MESSAGE_SIZE]);
 
         try {
-            producer.send(new byte[PulsarDecoder.MaxMessageSize + 1]);
+            producer.send(new byte[Commands.DEFAULT_MAX_MESSAGE_SIZE + 1]);
             fail("Should have thrown exception");
         } catch (PulsarClientException.InvalidMessageException e) {
             // OK
@@ -671,7 +671,7 @@ public void testSendBigMessageSizeButCompressed() throws Exception {
             .messageRoutingMode(MessageRoutingMode.SinglePartition)
             .compressionType(CompressionType.LZ4)
             .create();
-        producer.send(new byte[PulsarDecoder.MaxMessageSize + 1]);
+        producer.send(new byte[Commands.DEFAULT_MAX_MESSAGE_SIZE + 1]);
         producer.close();
 
         // (b) batch-msg with compression
@@ -680,7 +680,7 @@ public void testSendBigMessageSizeButCompressed() throws Exception {
             .messageRoutingMode(MessageRoutingMode.SinglePartition)
             .compressionType(CompressionType.LZ4)
             .create();
-        producer.send(new byte[PulsarDecoder.MaxMessageSize + 1]);
+        producer.send(new byte[Commands.DEFAULT_MAX_MESSAGE_SIZE + 1]);
         producer.close();
 
         // (c) non-batch msg without compression
@@ -690,7 +690,7 @@ public void testSendBigMessageSizeButCompressed() throws Exception {
             .compressionType(CompressionType.NONE)
             .create();
         try {
-            producer.send(new byte[PulsarDecoder.MaxMessageSize + 1]);
+            producer.send(new byte[Commands.DEFAULT_MAX_MESSAGE_SIZE + 1]);
             fail("Should have thrown exception");
         } catch (PulsarClientException.InvalidMessageException e) {
             // OK
@@ -704,7 +704,7 @@ public void testSendBigMessageSizeButCompressed() throws Exception {
             .messageRoutingMode(MessageRoutingMode.SinglePartition)
             .compressionType(CompressionType.LZ4).create();
         Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1").subscribe();
-        byte[] content = new byte[PulsarDecoder.MaxMessageSize + 10];
+        byte[] content = new byte[Commands.DEFAULT_MAX_MESSAGE_SIZE + 10];
         producer.send(content);
         assertEquals(consumer.receive().getData(), content);
         producer.close();
@@ -716,7 +716,7 @@ public void testSendBigMessageSizeButCompressed() throws Exception {
             .compressionType(CompressionType.NONE)
             .create();
         try {
-            producer.send(new byte[PulsarDecoder.MaxMessageSize + 1]);
+            producer.send(new byte[Commands.DEFAULT_MAX_MESSAGE_SIZE + 1]);
             fail("Should have thrown exception");
         } catch (PulsarClientException.InvalidMessageException e) {
             // OK

pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java

@@ -73,6 +73,7 @@
 import org.apache.pulsar.client.impl.ConsumerImpl;
 import org.apache.pulsar.client.impl.MessageIdImpl;
 import org.apache.pulsar.client.impl.TypedMessageBuilderImpl;
+import org.apache.pulsar.common.api.Commands;
 import org.apache.pulsar.common.api.PulsarDecoder;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.util.FutureUtil;
@@ -607,10 +608,10 @@ public void testSendBigMessageSize() throws Exception {
         Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
 
         // Messages are allowed up to MaxMessageSize
-        producer.newMessage().value(new byte[PulsarDecoder.MaxMessageSize]);
+        producer.newMessage().value(new byte[Commands.DEFAULT_MAX_MESSAGE_SIZE]);
 
         try {
-            producer.send(new byte[PulsarDecoder.MaxMessageSize + 1]);
+            producer.send(new byte[Commands.DEFAULT_MAX_MESSAGE_SIZE + 1]);
             fail("Should have thrown exception");
         } catch (PulsarClientException.InvalidMessageException e) {
             // OK

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageParserTest.java

@@ -34,6 +34,7 @@
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.api.Commands;
 import org.apache.pulsar.common.api.raw.MessageParser;
 import org.apache.pulsar.common.api.raw.RawMessage;
 import org.apache.pulsar.common.naming.TopicName;
@@ -91,7 +92,7 @@ public void testWithoutBatches() throws Exception {
                 MessageParser.parseMessage(topicName, entry.getLedgerId(), entry.getEntryId(), entry.getDataBuffer(),
                         (message) -> {
                             messages.add(message);
-                        });
+                        }, Commands.DEFAULT_MAX_MESSAGE_SIZE);
             } finally {
                 entry.release();
             }
@@ -133,7 +134,7 @@ public void testWithBatches() throws Exception {
             MessageParser.parseMessage(topicName, entry.getLedgerId(), entry.getEntryId(), entry.getDataBuffer(),
                     (message) -> {
                         messages.add(message);
-                    });
+                    }, Commands.DEFAULT_MAX_MESSAGE_SIZE);
         } finally {
             entry.release();
         }