diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java index 0b93ce27bfc812..53a6f26170c781 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java @@ -188,7 +188,7 @@ public interface ProducerBuilder extends Cloneable { * The purpose of this setting is to have an upper-limit on the number * of pending messages when publishing on a partitioned topic. * - *

Default is 50000. + *

Default is 0, disable the pending messages across partitions check. * *

If publishing at high rate over a topic with many partitions (especially when publishing messages without a * partitioning key), it might be beneficial to increase this parameter to allow for more pipelining within the @@ -198,6 +198,7 @@ public interface ProducerBuilder extends Cloneable { * max pending messages across all the partitions * @return the producer builder instance */ + @Deprecated ProducerBuilder maxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPartitions); /** diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java index 497b82a78c7f88..90a2fc8666a1ea 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java @@ -145,6 +145,7 @@ public ProducerBuilder maxPendingMessages(int maxPendingMessages) { return this; } + @Deprecated @Override public ProducerBuilder maxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPartitions) { conf.setMaxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java index f94bf5b057a30f..08fc096e660299 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java @@ -54,7 +54,7 @@ public class ProducerConfigurationData implements Serializable, Cloneable { public static final int DEFAULT_BATCHING_MAX_MESSAGES = 1000; public static final int DEFAULT_MAX_PENDING_MESSAGES = 0; - public static final int DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS = 50000; + public static final int DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS = 0; private String topicName = null; private String producerName = null; diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java index 21708c25f75370..dd62843c6d3235 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java @@ -346,7 +346,7 @@ public void testProducerBuilderImplWhenSendTimeoutPropertyIsNegative() { @Test(expectedExceptions = IllegalArgumentException.class) public void testProducerBuilderImplWhenMaxPendingMessagesAcrossPartitionsPropertyIsInvalid() { - producerBuilderImpl.maxPendingMessagesAcrossPartitions(999); + producerBuilderImpl.maxPendingMessagesAcrossPartitions(-1); } @Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = "maxPendingMessagesAcrossPartitions needs to be >= maxPendingMessages")