diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index 5428d557a90e5..5209e311e24a2 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/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, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java index 3ef104f82fdd9..8f291ea31f4ef 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java +++ b/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; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java index f15b9a3f8d70b..65207e6c67829 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java +++ b/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()) { 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 ff24caa2ce797..f3e8214ffe1c9 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 @@ -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 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)); } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index 6fe386488848f..b5b6fe97b2d97 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/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 */) { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MaxMessageSizeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MaxMessageSizeTest.java new file mode 100644 index 0000000000000..4fb3183ab5a39 --- /dev/null +++ b/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(); + + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java index 7192e7fa6dd6a..32c6fa8f5ecb3 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java +++ b/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 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 diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java index fecef7946d5c1..2040979f36954 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java +++ b/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 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 diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageParserTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageParserTest.java index 5ec6332a0ceef..1e0c45c2ac29c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageParserTest.java +++ b/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(); } 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 1a88afd0c4bf2..7a58357ef9720 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 @@ -30,6 +30,7 @@ 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.SslHandler; import io.netty.util.concurrent.Promise; @@ -48,6 +49,7 @@ import javax.net.ssl.SSLSession; +import lombok.Getter; import org.apache.commons.lang3.tuple.Pair; import org.apache.http.conn.ssl.DefaultHostnameVerifier; import org.apache.pulsar.PulsarVersion; @@ -117,6 +119,10 @@ public class ClientCnx extends PulsarHandler { .newUpdater(ClientCnx.class, "numberOfRejectRequests"); @SuppressWarnings("unused") private volatile int numberOfRejectRequests = 0; + + @Getter + private static int maxMessageSize = Commands.DEFAULT_MAX_MESSAGE_SIZE; + private final int maxNumberOfRejectedRequestPerConnection; private final int rejectedRequestResetTimeSec = 60; private final int protocolVersion; @@ -275,7 +281,15 @@ protected void handleConnected(CommandConnected connected) { } checkArgument(state == State.SentConnectFrame || state == State.Connecting); - + if (connected.hasMaxMessageSize()) { + if (log.isDebugEnabled()) { + log.debug("{} Connection has max message size setting, replace old frameDecoder with " + + "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)); + } if (log.isDebugEnabled()) { log.debug("{} Connection is ready", ctx.channel()); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java index 8909989e5ada3..b2421f246335c 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java @@ -66,7 +66,6 @@ import org.apache.pulsar.common.api.Commands; import org.apache.pulsar.common.api.EncryptionContext; import org.apache.pulsar.common.api.EncryptionContext.EncryptionKey; -import org.apache.pulsar.common.api.PulsarDecoder; import org.apache.pulsar.common.api.proto.PulsarApi; import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType; import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.ValidationError; @@ -1141,10 +1140,10 @@ private ByteBuf uncompressPayloadIfNeeded(MessageIdData messageId, MessageMetada CompressionCodec codec = CompressionCodecProvider.getCompressionCodec(compressionType); int uncompressedSize = msgMetadata.getUncompressedSize(); int payloadSize = payload.readableBytes(); - if (payloadSize > PulsarDecoder.MaxMessageSize) { + if (payloadSize > ClientCnx.getMaxMessageSize()) { // payload size is itself corrupted since it cannot be bigger than the MaxMessageSize log.error("[{}][{}] Got corrupted payload message size {} at {}", topic, subscription, payloadSize, - messageId); + messageId); discardCorruptedMessage(messageId, currentCnx, ValidationError.UncompressedSizeCorruption); return null; } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java index b4a0993b49c8c..3e32fb91d2e3e 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java @@ -61,7 +61,6 @@ import org.apache.pulsar.common.api.ByteBufPair; import org.apache.pulsar.common.api.Commands; import org.apache.pulsar.common.api.Commands.ChecksumType; -import org.apache.pulsar.common.api.PulsarDecoder; import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata; import org.apache.pulsar.common.api.proto.PulsarApi.ProtocolVersion; import org.apache.pulsar.common.compression.CompressionCodec; @@ -312,15 +311,14 @@ public void sendAsync(Message 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) { + if (compressedSize > ClientCnx.getMaxMessageSize()) { compressedPayload.release(); String compressedStr = (!isBatchMessagingEnabled() && conf.getCompressionType() != CompressionType.NONE) - ? "Compressed" - : ""; + ? "Compressed" + : ""; PulsarClientException.InvalidMessageException invalidMessageException = new PulsarClientException.InvalidMessageException( - format("%s Message payload size %d cannot exceed %d bytes", compressedStr, compressedSize, - PulsarDecoder.MaxMessageSize)); + format("%s Message payload size %d cannot exceed %d bytes", compressedStr, compressedSize, + ClientCnx.getMaxMessageSize())); callback.sendComplete(invalidMessageException); return; } @@ -1306,12 +1304,12 @@ private void batchMessageAndSend() { op = OpSendMsg.create(batchMessageContainer.messages, cmd, sequenceId, batchMessageContainer.firstCallback); - if (encryptedPayload.readableBytes() > PulsarDecoder.MaxMessageSize) { + if (encryptedPayload.readableBytes() > ClientCnx.getMaxMessageSize()) { cmd.release(); semaphore.release(numMessagesInBatch); if (op != null) { op.callback.sendComplete(new PulsarClientException.InvalidMessageException( - "Message size is bigger than " + PulsarDecoder.MaxMessageSize + " bytes")); + "Message size is bigger than " + ClientCnx.getMaxMessageSize() + " bytes")); } return; } 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 494cb1283264b..5d0f87273c50e 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 @@ -29,6 +29,7 @@ import org.apache.pulsar.client.api.AuthenticationDataProvider; import org.apache.pulsar.client.impl.conf.ClientConfigurationData; import org.apache.pulsar.common.api.ByteBufPair; +import org.apache.pulsar.common.api.Commands; import org.apache.pulsar.common.api.PulsarDecoder; import org.apache.pulsar.common.util.SecurityUtility; @@ -38,6 +39,7 @@ public class PulsarChannelInitializer extends ChannelInitializer private final Supplier clientCnxSupplier; private final SslContext sslCtx; + private final ClientConfigurationData conf; public PulsarChannelInitializer(ClientConfigurationData conf, Supplier clientCnxSupplier) throws Exception { @@ -57,6 +59,7 @@ public PulsarChannelInitializer(ClientConfigurationData conf, Supplier PulsarDecoder.MaxMessageSize) { + if (payloadSize > maxMessageSize) { // payload size is itself corrupted since it cannot be bigger than the MaxMessageSize log.error("[{}] Got corrupted payload message size {} at {}:{}", topic, payloadSize, ledgerId, entryId); diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto index d73e60190a316..11d69bee21c48 100644 --- a/pulsar-common/src/main/proto/PulsarApi.proto +++ b/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; } message CommandAuthResponse { diff --git a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/ServerConnection.java b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/ServerConnection.java index 0f334623272d6..b89a0285b9c86 100644 --- a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/ServerConnection.java +++ b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/ServerConnection.java @@ -36,6 +36,7 @@ import org.apache.pulsar.common.api.proto.PulsarApi.CommandPartitionedTopicMetadata; import org.apache.pulsar.common.api.proto.PulsarApi.ServerError; import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.discovery.service.server.ServiceConfig; import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/ServiceChannelInitializer.java b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/ServiceChannelInitializer.java index 212a7404c1c17..50d698d076612 100644 --- a/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/ServiceChannelInitializer.java +++ b/pulsar-discovery-service/src/main/java/org/apache/pulsar/discovery/service/ServiceChannelInitializer.java @@ -18,9 +18,8 @@ */ package org.apache.pulsar.discovery.service; -import org.apache.pulsar.common.api.PulsarDecoder; +import org.apache.pulsar.common.api.Commands; import org.apache.pulsar.common.util.NettySslContextBuilder; -import org.apache.pulsar.common.util.SslContextAutoRefreshBuilder; import org.apache.pulsar.discovery.service.server.ServiceConfig; import io.netty.channel.ChannelInitializer; @@ -63,7 +62,8 @@ protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(TLS_HANDLER, sslContext.newHandler(ch.alloc())); } } - ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(PulsarDecoder.MaxFrameSize, 0, 4, 0, 4)); + ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder( + Commands.DEFAULT_MAX_MESSAGE_SIZE + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0, 4)); ch.pipeline().addLast("handler", new ServerConnection(discoveryService)); } } 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 ffa4c2c838cb9..a8deb4425e175 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 @@ -21,7 +21,6 @@ import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkState; -import static java.nio.charset.StandardCharsets.UTF_8; import java.net.URI; import java.net.URISyntaxException; @@ -39,6 +38,7 @@ import org.apache.pulsar.common.api.PulsarDecoder; import org.apache.pulsar.common.api.proto.PulsarApi.CommandAuthChallenge; import org.apache.pulsar.common.api.proto.PulsarApi.CommandConnected; +import org.apache.pulsar.common.conf.InternalConfigurationData; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -98,8 +98,8 @@ protected void initChannel(SocketChannel ch) throws Exception { if (sslCtx != null) { ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc())); } - ch.pipeline().addLast("frameDecoder", - new LengthFieldBasedFrameDecoder(PulsarDecoder.MaxFrameSize, 0, 4, 0, 4)); + ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder( + Commands.DEFAULT_MAX_MESSAGE_SIZE + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0, 4)); ch.pipeline().addLast("proxyOutboundHandler", new ProxyBackendHandler(config, protocolVersion)); } }); @@ -259,7 +259,15 @@ protected void handleConnected(CommandConnected connected) { state = BackendState.HandshakeCompleted; - inboundChannel.writeAndFlush(Commands.newConnected(connected.getProtocolVersion())).addListener(future -> { + ChannelFuture channelFuture; + if (connected.hasMaxMessageSize()) { + channelFuture = inboundChannel.writeAndFlush( + Commands.newConnected(connected.getProtocolVersion(), connected.getMaxMessageSize())); + } else { + channelFuture = inboundChannel.writeAndFlush(Commands.newConnected(connected.getProtocolVersion())); + } + + channelFuture.addListener(future -> { if (log.isDebugEnabled()) { log.debug("[{}] [{}] Removing decoder from pipeline", inboundChannel, outboundChannel); } @@ -270,8 +278,34 @@ protected void handleConnected(CommandConnected connected) { } else { // Enable parsing feature, proxyLogLevel(1 or 2) // Add parser handler - inboundChannel.pipeline().addBefore("handler" , "inboundParser" , new ParserProxyHandler(inboundChannel , ParserProxyHandler.FRONTEND_CONN)); - outboundChannel.pipeline().addBefore("proxyOutboundHandler" , "outboundParser" , new ParserProxyHandler(outboundChannel , ParserProxyHandler.BACKEND_CONN)); + 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)); + + inboundChannel.pipeline().addBefore("handler", "inboundParser", + new ParserProxyHandler(inboundChannel, + ParserProxyHandler.FRONTEND_CONN, + connected.getMaxMessageSize())); + outboundChannel.pipeline().addBefore("proxyOutboundHandler", "outboundParser", + new ParserProxyHandler(outboundChannel, + ParserProxyHandler.BACKEND_CONN, + connected.getMaxMessageSize())); + } else { + inboundChannel.pipeline().addBefore("handler", "inboundParser", + new ParserProxyHandler(inboundChannel, + ParserProxyHandler.FRONTEND_CONN, + Commands.DEFAULT_MAX_MESSAGE_SIZE)); + outboundChannel.pipeline().addBefore("proxyOutboundHandler", "outboundParser", + new ParserProxyHandler(outboundChannel, + ParserProxyHandler.BACKEND_CONN, + Commands.DEFAULT_MAX_MESSAGE_SIZE)); + } } // Start reading from both connections inboundChannel.read(); diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java index 040a8349b4f88..8b4fe64545347 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ParserProxyHandler.java @@ -51,15 +51,18 @@ public class ParserProxyHandler extends ChannelInboundHandlerAdapter { private String connType; + private int maxMessageSize; + //producerid+channelid as key //or consumerid+channelid as key private static Map producerHashMap = new ConcurrentHashMap<>(); private static Map consumerHashMap = new ConcurrentHashMap<>(); - public ParserProxyHandler(Channel channel, String type){ + public ParserProxyHandler(Channel channel, String type, int maxMessageSize){ this.channel = channel; this.connType=type; + this.maxMessageSize = maxMessageSize; } private void logging(Channel conn, PulsarApi.BaseCommand.Type cmdtype, String info, List messages) throws Exception{ @@ -118,7 +121,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) { MessageParser.parseMessage(topicName, -1L, -1L,buffer,(message) -> { messages.add(message); - }); + }, maxMessageSize); logging(ctx.channel() , cmd.getType() , "" , messages); break; @@ -138,7 +141,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) { MessageParser.parseMessage(topicName, -1L, -1L,buffer,(message) -> { messages.add(message); - }); + }, maxMessageSize); logging(ctx.channel() , cmd.getType() , "" , messages); 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 b03afbfdf4460..9be109c2027b1 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 @@ -22,10 +22,9 @@ import org.apache.pulsar.client.api.AuthenticationDataProvider; import org.apache.pulsar.client.api.AuthenticationFactory; -import org.apache.pulsar.common.api.PulsarDecoder; +import org.apache.pulsar.common.api.Commands; import org.apache.pulsar.common.util.ClientSslContextRefresher; import org.apache.pulsar.common.util.NettySslContextBuilder; -import org.apache.pulsar.common.util.SslContextAutoRefreshBuilder; import io.netty.channel.ChannelInitializer; import io.netty.channel.socket.SocketChannel; @@ -85,7 +84,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( + Commands.DEFAULT_MAX_MESSAGE_SIZE + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0, 4)); ch.pipeline().addLast("handler", new ProxyConnection(proxyService, clientSslCtxRefresher == null ? null : clientSslCtxRefresher.get())); } diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorConfig.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorConfig.java index 23992334ca149..d36c90dbc56fd 100644 --- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorConfig.java +++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorConfig.java @@ -23,6 +23,7 @@ import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.bookkeeper.stats.NullStatsProvider; +import org.apache.pulsar.common.api.Commands; import javax.validation.constraints.NotNull; import java.io.IOException; @@ -37,6 +38,7 @@ public class PulsarConnectorConfig implements AutoCloseable { private int targetNumSplits = 2; private int maxSplitMessageQueueSize = 10000; private int maxSplitEntryQueueSize = 1000; + private int maxMessageSize = Commands.DEFAULT_MAX_MESSAGE_SIZE; private String statsProvider = NullStatsProvider.class.getName(); private Map statsProviderConfigs = new HashMap<>(); @@ -59,6 +61,16 @@ public PulsarConnectorConfig setBrokerServiceUrl(String brokerServiceUrl) { return this; } + @Config("pulsar.max-message-size") + public PulsarConnectorConfig setMaxMessageSize(int maxMessageSize) { + this.maxMessageSize = maxMessageSize; + return this; + } + + public int getMaxMessageSize() { + return this.maxMessageSize; + } + @NotNull public String getZookeeperUri() { return this.zookeeperUri; diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java index 4ffbd2f2e80a9..09f3fa624145f 100644 --- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java +++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java @@ -138,6 +138,7 @@ private void initialize(List columnHandles, PulsarSplit puls pulsarSplit.getTableName()); this.metricsTracker = pulsarConnectorMetricsTracker; this.readOffloaded = pulsarConnectorConfig.getManagedLedgerOffloadDriver() != null; + this.pulsarConnectorConfig = pulsarConnectorConfig; Schema schema = PulsarConnectorUtils.parseSchema(pulsarSplit.getSchema()); @@ -260,7 +261,7 @@ public void accept(Entry entry) { } catch (InterruptedException e) { //no-op } - }); + }, pulsarConnectorConfig.getMaxMessageSize()); } catch (IOException e) { log.error(e, "Failed to parse message from pulsar topic %s", topicName.toString()); throw new RuntimeException(e);