diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerQueueSizeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerQueueSizeTest.java index ba28447eae804..cc4b53d101705 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerQueueSizeTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerQueueSizeTest.java @@ -18,16 +18,37 @@ */ package org.apache.pulsar.client.api; +import static org.assertj.core.api.Assertions.assertThat; import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; import lombok.Cleanup; import org.apache.pulsar.broker.service.SharedPulsarBaseTest; +import org.apache.pulsar.client.impl.ProducerBase; +import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; public class ProducerQueueSizeTest extends SharedPulsarBaseTest { + /** + * The bounds {@code PulsarClientImpl} falls back to when the client memory limit is disabled. + * Duplicated here on purpose: these are a documented client default, so a change to them should + * break a test rather than pass silently. + */ + private static final int NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES = 1000; + private static final int NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS = 50000; + + private static ProducerConfigurationData confOf(Producer producer) { + return ((ProducerBase) producer).getConfiguration(); + } + + @DataProvider(name = "partitioned") + public Object[][] partitioned() { + return new Object[][]{{Boolean.FALSE}, {Boolean.TRUE}}; + } + @DataProvider(name = "matrix") public Object[][] matrix() { return new Object[][]{ @@ -72,4 +93,277 @@ public void testRemoveMaxQueueLimit(boolean blockIfQueueFull, boolean partitione f.get(); } } + + /** + * A client with the memory limit disabled has no byte-based backpressure, so producers must fall + * back to a bounded pending-message queue. This has to hold for the no-argument + * {@code newProducer()} overload as well, not just {@code newProducer(Schema)}. + */ + @Test + public void testNoArgNewProducerIsBoundedWhenMemoryLimitDisabled() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer().topic(newTopicName()).create(); + + assertThat(confOf(producer).getMaxPendingMessages()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); + } + + /** + * The fallback is a default, not a floor. An application that asks for no message-count limit at + * all still gets it, by passing 0 explicitly. This is what keeps 0 a usable value rather than an + * alias for "unset". + * + *

A single {@code maxPendingMessages(0)} has to be enough whatever the topic's shape: filling + * in the across-partitions budget would put a per-partition limit back on a partitioned topic. + */ + @Test(dataProvider = "partitioned") + public void testExplicitZeroDisablesTheBoundWhenMemoryLimitDisabled(boolean partitioned) throws Exception { + String topic = newTopicName(); + if (partitioned) { + admin.topics().createPartitionedTopic(topic, 10); + } + + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer(Schema.BYTES) + .topic(topic) + .maxPendingMessages(0) + .create(); + + assertThat(confOf(producer).getMaxPendingMessages()).isZero(); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()).isZero(); + } + + /** + * Mirror of the above: disabling only the across-partitions budget must not take the per-producer + * default down with it. Filling in that default would otherwise be capped by a budget of 0. + */ + @SuppressWarnings("deprecation") + @Test + public void testExplicitZeroAcrossPartitionsKeepsThePerProducerDefault() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer() + .topic(newTopicName()) + .maxPendingMessagesAcrossPartitions(0) + .create(); + + assertThat(confOf(producer).getMaxPendingMessages()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()).isZero(); + } + + /** + * {@code loadConf} is the other way an application configures a limit. A limit present in the map + * counts as configured, including a 0, even though {@code loadConf} rebuilds the configuration + * object and so cannot carry any marker on it. + */ + @Test + public void testLoadConfZeroDisablesTheBoundWhenMemoryLimitDisabled() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer() + .topic(newTopicName()) + .loadConf(Map.of("maxPendingMessages", 0)) + .create(); + + assertThat(confOf(producer).getMaxPendingMessages()).isZero(); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()).isZero(); + } + + /** + * A {@code loadConf} that does not mention the limits leaves them unconfigured, so the defaults + * still apply. Pins that rebuilding the configuration is not mistaken for configuring it. + */ + @Test + public void testLoadConfWithoutTheLimitsKeepsTheDefaults() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer() + .topic(newTopicName()) + .loadConf(Map.of("producerName", "loadConfWithoutLimits")) + .create(); + + assertThat(confOf(producer).getMaxPendingMessages()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); + } + + /** + * A cloned builder has to keep knowing which limits were configured, or the clone would silently + * get the defaults back. + */ + @Test + public void testCloneKeepsAnExplicitlyDisabledBound() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + ProducerBuilder builder = client.newProducer().maxPendingMessages(0); + + @Cleanup + Producer producer = builder.clone().topic(newTopicName()).create(); + + assertThat(confOf(producer).getMaxPendingMessages()).isZero(); + } + + /** + * {@code maxPendingMessagesAcrossPartitions} must be {@code >= maxPendingMessages}. Filling in + * the across-partitions fallback must therefore never lower it below an explicitly configured + * per-partition limit, which would fail producer creation. + */ + @Test + public void testExplicitMaxPendingMessagesAboveTheFallbackDoesNotFailCreation() throws Exception { + int maxPendingMessages = NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS + 10_000; + + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer() + .topic(newTopicName()) + .maxPendingMessages(maxPendingMessages) + .create(); + + assertThat(confOf(producer).getMaxPendingMessages()).isEqualTo(maxPendingMessages); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()) + .isGreaterThanOrEqualTo(maxPendingMessages); + } + + /** + * Filling in the fallback must not write it back into the builder's own configuration. The + * builder stays reusable, and a limit set on it afterwards is still validated against what the + * caller configured rather than against a filled-in default. + */ + @SuppressWarnings("deprecation") + @Test + public void testFallbackLeavesTheBuilderReusable() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + ProducerBuilder builder = client.newProducer(); + + @Cleanup + Producer first = builder.topic(newTopicName()).create(); + assertThat(confOf(first).getMaxPendingMessages()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES); + + // Rejected if creating the first producer had left the fallback in the builder, since the + // across-partitions limit has to be >= maxPendingMessages. + @Cleanup + Producer second = builder.topic(newTopicName()) + .maxPendingMessagesAcrossPartitions(500) + .create(); + assertThat(confOf(second).getMaxPendingMessages()).isEqualTo(500); + } + + /** + * The fallback has to reach partitioned producers too, where the per-partition queue is derived + * from the across-partitions budget. + */ + @Test + public void testPartitionedProducerIsBoundedWhenMemoryLimitDisabled() throws Exception { + String topic = newTopicName(); + admin.topics().createPartitionedTopic(topic, 10); + + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer().topic(topic).create(); + + // The budget spread over 10 partitions is well above the per-producer default, so each + // partition keeps the full default. + assertThat(confOf(producer).getMaxPendingMessages()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); + } + + /** + * The across-partitions limit is a budget shared by every partition, so the per-producer + * fallback must be capped by it. Otherwise the fallback would exceed an explicitly configured + * budget, which producer creation rejects. + */ + @SuppressWarnings("deprecation") + @Test + public void testExplicitAcrossPartitionsLimitCapsTheFallback() throws Exception { + int maxPendingMessagesAcrossPartitions = 500; + + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer() + .topic(newTopicName()) + .maxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions) + .create(); + + assertThat(confOf(producer).getMaxPendingMessages()) + .isEqualTo(maxPendingMessagesAcrossPartitions); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()) + .isEqualTo(maxPendingMessagesAcrossPartitions); + } + + /** + * The fallback only exists to replace the missing byte-based backpressure. When a memory limit + * is configured, an unset pending-message limit keeps meaning "no message-count limit". + */ + @Test + public void testMemoryLimitedClientKeepsUnboundedPendingMessages() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(64, SizeUnit.MEGA_BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer().topic(newTopicName()).create(); + + assertThat(confOf(producer).getMaxPendingMessages()).isZero(); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()).isZero(); + } } 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 7b35432da1d6a..4e37bb4d19809 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 @@ -174,7 +174,15 @@ public interface ProducerBuilder extends Cloneable { * the client application. Until the producer gets a successful acknowledgment back from the broker, * it will keep in memory (direct memory pool) all the messages in the pending queue. * - *

Default is 0, which disables the pending messages check. + *

Default is 0, which disables the pending messages check. Disabling it only removes the + * message-count limit; the memory the pending queue may hold is then bounded by the client + * memory limit ({@link ClientBuilder#memoryLimit(long, SizeUnit)}) instead. + * + *

On a client whose memory limit is disabled there would be no backpressure left at all, so a + * producer that does not configure this setting falls back to a default queue size of 1000 rather + * than buffering without limit. Calling this method always wins over that default, so passing 0 + * explicitly is how an application asks for a producer with no message-count limit, on a + * partitioned topic as well. * * @param maxPendingMessages * the max size of the pending messages queue for the producer @@ -190,7 +198,10 @@ 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 0, which disables the pending messages across partitions check. + *

Default is 0, which disables the pending messages across partitions check. As with + * {@link #maxPendingMessages(int)}, a producer that does not configure this setting on a client + * whose memory limit is disabled falls back to a default budget of 50000 instead, since no + * backpressure would otherwise be left, and calling this method always wins over that default. * *

If publishing at a 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 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 9242cfd6a08cf..11915140ceab9 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 @@ -56,6 +56,14 @@ public class ProducerBuilderImpl implements ProducerBuilder { private ProducerConfigurationData conf; private Schema schema; private List interceptorList; + /** + * Whether the application configured the pending-message limits. Their unset value is 0, which is + * also a meaningful explicit value ("no message-count limit"), so the configuration alone cannot + * tell the two apart. See + * {@link PulsarClientImpl#applyNoMemoryLimitProducerDefaults(ProducerConfigurationData, boolean, boolean)}. + */ + private boolean maxPendingMessagesConfigured; + private boolean maxPendingMessagesAcrossPartitionsConfigured; public ProducerBuilderImpl(PulsarClientImpl client, Schema schema) { this(client, new ProducerConfigurationData(), schema); @@ -78,7 +86,10 @@ public ProducerBuilder schema(Schema schema) { @Override public ProducerBuilder clone() { - return new ProducerBuilderImpl<>(client, conf.clone(), schema); + ProducerBuilderImpl copy = new ProducerBuilderImpl<>(client, conf.clone(), schema); + copy.maxPendingMessagesConfigured = maxPendingMessagesConfigured; + copy.maxPendingMessagesAcrossPartitionsConfigured = maxPendingMessagesAcrossPartitionsConfigured; + return copy; } @Override @@ -120,15 +131,23 @@ public CompletableFuture> createAsync() { client.instrumentProvider())); } + ProducerConfigurationData producerConf = client.applyNoMemoryLimitProducerDefaults(conf, + maxPendingMessagesConfigured, maxPendingMessagesAcrossPartitionsConfigured); + return effectiveInterceptors == null || effectiveInterceptors.size() == 0 - ? client.createProducerAsync(conf, schema, null) - : client.createProducerAsync(conf, schema, new ProducerInterceptors(effectiveInterceptors)); + ? client.createProducerAsync(producerConf, schema, null) + : client.createProducerAsync(producerConf, schema, new ProducerInterceptors(effectiveInterceptors)); } @Override public ProducerBuilder loadConf(Map config) { conf = ConfigurationDataUtils.loadData( config, conf, ProducerConfigurationData.class); + // A limit present in the map was configured by the application, even when its value is the + // same as the unset one. loadData builds a new configuration instance, so this cannot be + // tracked in the configuration itself. + maxPendingMessagesConfigured |= config.containsKey("maxPendingMessages"); + maxPendingMessagesAcrossPartitionsConfigured |= config.containsKey("maxPendingMessagesAcrossPartitions"); return this; } @@ -154,6 +173,7 @@ public ProducerBuilder sendTimeout(int sendTimeout, @NonNull TimeUnit unit) { @Override public ProducerBuilder maxPendingMessages(int maxPendingMessages) { conf.setMaxPendingMessages(maxPendingMessages); + maxPendingMessagesConfigured = true; return this; } @@ -161,6 +181,7 @@ public ProducerBuilder maxPendingMessages(int maxPendingMessages) { @Override public ProducerBuilder maxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPartitions) { conf.setMaxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions); + maxPendingMessagesAcrossPartitionsConfigured = true; return this; } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java index 873069c1c930f..221c75436c063 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java @@ -609,17 +609,9 @@ public ProducerBuilder newProducer() { return new ProducerBuilderImpl<>(this, Schema.BYTES); } - @SuppressWarnings("deprecation") @Override public ProducerBuilder newProducer(Schema schema) { - ProducerBuilderImpl producerBuilder = new ProducerBuilderImpl<>(this, schema); - if (!memoryLimitController.isMemoryLimited()) { - // set default limits for producers when memory limit controller is disabled - producerBuilder.maxPendingMessages(NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES); - producerBuilder.maxPendingMessagesAcrossPartitions( - NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); - } - return producerBuilder; + return new ProducerBuilderImpl<>(this, schema); } @Override @@ -714,6 +706,80 @@ public CompletableFuture> createProducerAsync(ProducerConfigurat } + /** + * Apply the default pending-message limits a producer gets when this client has no memory limit. + * + *

The client memory limit is a producer's primary backpressure: it bounds the memory held by + * messages that have been queued but not yet acknowledged by the broker. When it is disabled + * there is nothing left to bound that queue, so producers fall back to the pre-PIP-120 + * message-count defaults rather than buffering without any limit at all. + * + *

These are defaults, not a floor. A limit the application configured is always kept — including + * an explicit {@code 0}, which is how an application asks for no message-count limit at all. Only a + * limit that was never configured is filled in, which is why the caller passes in what it saw + * rather than letting this method infer it: {@code 0} is both the unset value and a meaningful + * explicit one. + * + *

Note that on a partitioned topic a filled-in across-partitions budget is still divided between + * the partitions afterwards, which can lower an explicitly configured per-producer limit. + * + *

Called by {@link ProducerBuilderImpl}, which is what knows whether a limit was configured. The + * V5 client builds its segment producers through {@link #createSegmentProducerAsync} instead, and + * deliberately gets no defaults here: it exposes no pending-message setting at all, so its client + * memory limit is the only backpressure it has and the only thing an application can turn off. + * + * @param conf the requested producer configuration + * @param maxPendingMessagesConfigured whether the application configured {@code maxPendingMessages} + * @param maxPendingMessagesAcrossPartitionsConfigured whether the application configured + * {@code maxPendingMessagesAcrossPartitions} + * @return the configuration to create the producer with; a resolved copy when a default applies, + * otherwise {@code conf} unchanged + */ + public ProducerConfigurationData applyNoMemoryLimitProducerDefaults(ProducerConfigurationData conf, + boolean maxPendingMessagesConfigured, boolean maxPendingMessagesAcrossPartitionsConfigured) { + // A limit that is already positive was configured by definition, whichever way the + // configuration was populated. The flags only tell an explicit 0 apart from an unset one. + maxPendingMessagesConfigured |= conf.getMaxPendingMessages() > 0; + maxPendingMessagesAcrossPartitionsConfigured |= conf.getMaxPendingMessagesAcrossPartitions() > 0; + if ((maxPendingMessagesConfigured && maxPendingMessagesAcrossPartitionsConfigured) + || memoryLimitController.isMemoryLimited()) { + return conf; + } + int maxPendingMessages = maxPendingMessagesConfigured + ? conf.getMaxPendingMessages() + : NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES; + final int maxPendingMessagesAcrossPartitions; + if (maxPendingMessagesAcrossPartitionsConfigured) { + maxPendingMessagesAcrossPartitions = conf.getMaxPendingMessagesAcrossPartitions(); + } else if (maxPendingMessages == 0) { + // The application configured no per-producer limit. Filling in a partitions budget would + // put one back, because a partitioned producer derives its per-partition limit from it, so + // a single maxPendingMessages(0) is enough to ask for a producer with no message-count + // limit whatever the topic's shape. + maxPendingMessagesAcrossPartitions = 0; + } else { + maxPendingMessagesAcrossPartitions = + Math.max(maxPendingMessages, NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); + } + if (maxPendingMessagesAcrossPartitions > 0) { + // The across-partitions limit is a budget shared by every partition, so a single producer's + // queue can never exceed it. A configured 0 means there is no such budget and is left + // alone, rather than capping every producer at zero. + maxPendingMessages = Math.min(maxPendingMessages, maxPendingMessagesAcrossPartitions); + } + + // Resolve on a copy: the builder hands over its own configuration instance, so filling in a + // limit here would otherwise leak into the next producer built from the same builder. + ProducerConfigurationData resolved = conf.clone(); + resolved.setMaxPendingMessages(maxPendingMessages); + resolved.setMaxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions); + log.debug().attr("topic", conf.getTopicName()) + .attr("maxPendingMessages", maxPendingMessages) + .attr("maxPendingMessagesAcrossPartitions", maxPendingMessagesAcrossPartitions) + .log("Client memory limit is disabled, applying default producer pending message limits"); + return resolved; + } + @SuppressWarnings("unchecked") public CompletableFuture reloadSchemaForAutoProduceProducer(String topic, AutoProduceBytesSchema autoSchema) { return lookup.getSchema(TopicName.get(topic)).thenAccept(schemaInfoOptional -> { 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 601cf78c8b893..02c25bece9053 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 @@ -252,9 +252,17 @@ public void setMaxPendingMessages(int maxPendingMessages) { this.maxPendingMessages = maxPendingMessages; } + /** + * The across-partitions budget used to be rejected when it was below {@link #maxPendingMessages}, + * which made the two setters order-dependent: it depended on which of them had been called first, + * and it made {@code loadConf} fail outright for any positive {@code maxPendingMessages}, since + * that replays every property through the setters in an order the caller does not control. The + * relationship is enforced where it is used instead — {@code PartitionedProducerImpl} lowers the + * per-partition limit to the share of the budget when a budget is set. + */ public void setMaxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPartitions) { - checkArgument(maxPendingMessagesAcrossPartitions >= maxPendingMessages, - "maxPendingMessagesAcrossPartitions needs to be >= maxPendingMessages"); + checkArgument(maxPendingMessagesAcrossPartitions >= 0, + "maxPendingMessagesAcrossPartitions needs to be >= 0"); this.maxPendingMessagesAcrossPartitions = maxPendingMessagesAcrossPartitions; } 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 7554135943194..bdae20c1ed22e 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 @@ -19,10 +19,13 @@ package org.apache.pulsar.client.impl; import static org.mockito.Mockito.any; +import static org.mockito.Mockito.anyBoolean; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; @@ -63,6 +66,11 @@ public void setup() { producerBuilderImpl = new ProducerBuilderImpl<>(client, Schema.BYTES); when(client.newProducer()).thenReturn(producerBuilderImpl); + // The builder asks the client to fill in the pending-message defaults before creating the + // producer; on a mock that would otherwise hand back a null configuration. + when(client.applyNoMemoryLimitProducerDefaults(any(ProducerConfigurationData.class), anyBoolean(), + anyBoolean())).thenAnswer(invocation -> invocation.getArgument(0)); + doReturn(CompletableFuture.completedFuture(producer)) .when(client).createProducerAsync( any(ProducerConfigurationData.class), any(), eq(null)); @@ -119,6 +127,23 @@ public void testProducerBuilderImplWhenMessageRoutingModeIsRoundRobinPartition() assertNotNull(producer); } + /** + * {@code loadConf} rebuilds the configuration by replaying every property through the public + * setters, and {@code setMaxPendingMessagesAcrossPartitions} rejects a value below + * {@code maxPendingMessages}. Pins that loading a positive limit does not trip that check on the + * across-partitions property that comes with it, and that the limit is recorded as configured. + */ + @SuppressWarnings("deprecation") + @Test + public void testLoadConfWithAPositiveMaxPendingMessages() { + producerBuilderImpl = new ProducerBuilderImpl<>(client, Schema.BYTES); + producerBuilderImpl.loadConf(Map.of("maxPendingMessages", 5000)); + + assertEquals(producerBuilderImpl.getConf().getMaxPendingMessages(), 5000); + assertTrue(producerBuilderImpl.isMaxPendingMessagesConfigured()); + assertFalse(producerBuilderImpl.isMaxPendingMessagesAcrossPartitionsConfigured()); + } + @Test public void testProducerBuilderImplWhenMessageRoutingIsSetImplicitly() throws PulsarClientException { producerBuilderImpl = new ProducerBuilderImpl<>(client, Schema.BYTES); @@ -378,11 +403,25 @@ public void testProducerBuilderImplWhenMaxPendingMessagesAcrossPartitionsPropert @SuppressWarnings("deprecation") @Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = - "maxPendingMessagesAcrossPartitions needs to be >= maxPendingMessages") + "maxPendingMessagesAcrossPartitions needs to be >= 0") public void testProducerBuilderImplWhenMaxPendingMessagesAcrossPartitionsPropertyIsInvalidErrorMessages() { producerBuilderImpl.maxPendingMessagesAcrossPartitions(-1); } + /** + * The across-partitions budget is allowed to be below {@code maxPendingMessages}: it is a budget + * shared by every partition, and the per-partition limit is lowered to its share where it is used. + * Rejecting it here made the two setters order-dependent. + */ + @SuppressWarnings("deprecation") + @Test + public void testAcrossPartitionsLimitBelowMaxPendingMessagesIsAccepted() { + producerBuilderImpl = new ProducerBuilderImpl<>(client, Schema.BYTES); + producerBuilderImpl.maxPendingMessages(1000).maxPendingMessagesAcrossPartitions(500); + + assertEquals(producerBuilderImpl.getConf().getMaxPendingMessagesAcrossPartitions(), 500); + } + @SuppressWarnings("deprecation") @Test public void testProducerBuilderImplWhenNumericPropertiesAreValid() { diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java index f20fc07a5907d..d7fd8ad19979a 100644 --- a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java +++ b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java @@ -20,6 +20,7 @@ import static java.nio.charset.StandardCharsets.UTF_8; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.same; @@ -111,6 +112,10 @@ public void setup() throws PulsarClientException { when(client.newProducer()).thenAnswer(invocation -> new ProducerBuilderImpl<>(client, Schema.BYTES)); when(client.newProducer(any())).thenAnswer( invocation -> new ProducerBuilderImpl<>(client, invocation.getArgument(0))); + // The builder asks the client to fill in the pending-message defaults before creating the + // producer; on a mock that would otherwise hand back a null configuration. + when(client.applyNoMemoryLimitProducerDefaults(any(ProducerConfigurationData.class), anyBoolean(), + anyBoolean())).thenAnswer(invocation -> invocation.getArgument(0)); when(client.createProducerAsync(any(ProducerConfigurationData.class), any(), any())) .thenReturn(CompletableFuture.completedFuture(producer)); when(client.getSchema(anyString())).thenReturn(CompletableFuture.completedFuture(Optional.empty()));