You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/08/19 15:50:45 UTC
(pulsar) 02/02: [fix][client] Divide the across-partitions budget only when it was set explicitly (#26384)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a742f6f82dc3464b5299463f46d6e1fa7678c617
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Aug 19 18:17:58 2026 +0300
[fix][client] Divide the across-partitions budget only when it was set explicitly (#26384)
(cherry picked from commit 3be684acf3d81c4ff34bbd6dcd4c2cb41539f9b3)
---
.../pulsar/client/api/ProducerQueueSizeTest.java | 103 +++++++++++++++++++++
.../client/impl/PartitionedProducerImpl.java | 21 +++--
.../pulsar/client/impl/ProducerBuilderImpl.java | 34 +++----
.../pulsar/client/impl/PulsarClientImpl.java | 43 +++++----
.../impl/conf/ProducerConfigurationData.java | 26 ++++++
.../client/impl/PartitionedProducerImplTest.java | 11 ++-
.../client/impl/ProducerBuilderImplTest.java | 9 +-
.../pulsar/functions/instance/ContextImplTest.java | 5 +-
.../pulsar/testclient/PerformanceProducerTest.java | 4 +-
9 files changed, 197 insertions(+), 59 deletions(-)
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 abfe5b6c8c22..d8f34c564ecf 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
@@ -365,4 +365,107 @@ public class ProducerQueueSizeTest extends SharedPulsarBaseTest {
assertThat(confOf(producer).getMaxPendingMessages()).isZero();
assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()).isZero();
}
+
+ /**
+ * A budget the application asked for is what gets divided between the partitions. Pins that the
+ * per-producer default filled in alongside it does not win the division and cap every partition at
+ * that default instead of at its share of the budget.
+ */
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testExplicitAcrossPartitionsBudgetIsDividedBetweenThePartitions() throws Exception {
+ String topic = newTopicName();
+ admin.topics().createPartitionedTopic(topic, 10);
+
+ @Cleanup
+ PulsarClient client = PulsarClient.builder()
+ .serviceUrl(getWebServiceUrl())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ @Cleanup
+ Producer<byte[]> producer = client.newProducer()
+ .topic(topic)
+ .maxPendingMessagesAcrossPartitions(60_000)
+ .create();
+
+ assertThat(confOf(producer).getMaxPendingMessages()).isEqualTo(6_000);
+ }
+
+ /**
+ * The mirror image: a budget that was only filled in as a default must not be divided, because
+ * dividing it would lower a per-producer limit the application did ask for.
+ */
+ @Test
+ public void testFilledInBudgetDoesNotLowerAnExplicitPerProducerLimit() throws Exception {
+ String topic = newTopicName();
+ admin.topics().createPartitionedTopic(topic, 10);
+
+ @Cleanup
+ PulsarClient client = PulsarClient.builder()
+ .serviceUrl(getWebServiceUrl())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ @Cleanup
+ Producer<byte[]> producer = client.newProducer()
+ .topic(topic)
+ .maxPendingMessages(60_000)
+ .create();
+
+ assertThat(confOf(producer).getMaxPendingMessages()).isEqualTo(60_000);
+ }
+
+ /**
+ * A budget smaller than the partition count divides to zero, which used to remove the queue bound
+ * altogether — asking for a tighter budget made the producer unbounded. Each partition keeps the
+ * smallest possible queue instead.
+ */
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testBudgetSmallerThanThePartitionCountStillBoundsEachPartition() throws Exception {
+ String topic = newTopicName();
+ admin.topics().createPartitionedTopic(topic, 10);
+
+ @Cleanup
+ PulsarClient client = PulsarClient.builder()
+ .serviceUrl(getWebServiceUrl())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ @Cleanup
+ Producer<byte[]> producer = client.newProducer()
+ .topic(topic)
+ .maxPendingMessagesAcrossPartitions(5)
+ .create();
+
+ assertThat(confOf(producer).getMaxPendingMessages()).isEqualTo(1);
+ }
+
+ /**
+ * An explicit {@code 0} means "no message-count limit" and stays that way on a partitioned topic,
+ * even next to an across-partitions budget. Reading the unset value instead of the marker used to
+ * overwrite it with the budget's per-partition share.
+ */
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testExplicitZeroIsKeptAlongsideAnAcrossPartitionsBudget() throws Exception {
+ String topic = newTopicName();
+ admin.topics().createPartitionedTopic(topic, 10);
+
+ @Cleanup
+ PulsarClient client = PulsarClient.builder()
+ .serviceUrl(getWebServiceUrl())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ @Cleanup
+ Producer<byte[]> producer = client.newProducer()
+ .topic(topic)
+ .maxPendingMessages(0)
+ .maxPendingMessagesAcrossPartitions(60_000)
+ .create();
+
+ assertThat(confOf(producer).getMaxPendingMessages()).isZero();
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java
index c428919663c5..fa458605a4cd 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java
@@ -19,8 +19,6 @@
package org.apache.pulsar.client.impl;
import static com.google.common.base.Preconditions.checkArgument;
-import static org.apache.pulsar.client.impl.conf.ProducerConfigurationData.DEFAULT_MAX_PENDING_MESSAGES;
-import static org.apache.pulsar.client.impl.conf.ProducerConfigurationData.DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS;
import com.google.common.annotations.VisibleForTesting;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
@@ -84,14 +82,23 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
? new PartitionedTopicProducerStatsRecorderImpl()
: null;
+ // The across-partitions budget is a total shared by every partition, so it is divided between
+ // them here, where the partition count is finally known. Both limits are read through their
+ // "configured" markers rather than by comparing against their unset value, because that value
+ // is 0 for both and 0 is also a meaningful explicit setting.
// MaxPendingMessagesAcrossPartitions doesn't support partial partition such as SinglePartition correctly
int maxPendingMessages = conf.getMaxPendingMessages();
int maxPendingMessagesAcrossPartitions = conf.getMaxPendingMessagesAcrossPartitions();
- if (maxPendingMessagesAcrossPartitions != DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS) {
- int maxPendingMsgsForOnePartition = maxPendingMessagesAcrossPartitions / numPartitions;
- maxPendingMessages = (maxPendingMessages == DEFAULT_MAX_PENDING_MESSAGES)
- ? maxPendingMsgsForOnePartition
- : Math.min(maxPendingMessages, maxPendingMsgsForOnePartition);
+ if (conf.isMaxPendingMessagesAcrossPartitionsConfigured() && maxPendingMessagesAcrossPartitions > 0) {
+ // Never divide down to 0: a budget smaller than the partition count still asks for the
+ // smallest possible queue, not for the queue bound to be removed altogether.
+ int maxPendingMsgsForOnePartition =
+ Math.max(1, maxPendingMessagesAcrossPartitions / numPartitions);
+ // A per-producer limit the application set wins, including an explicit 0 ("no message-count
+ // limit"). Only a limit it left unset — or one filled in as a default — adopts the share.
+ maxPendingMessages = conf.isMaxPendingMessagesConfigured()
+ ? Math.min(maxPendingMessages, maxPendingMsgsForOnePartition)
+ : maxPendingMsgsForOnePartition;
conf.setMaxPendingMessages(maxPendingMessages);
}
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 c043ef57b0cb..d12d23d993ef 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,14 +56,6 @@ public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
private ProducerConfigurationData conf;
private Schema<T> schema;
private List<ProducerInterceptor> 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<T> schema) {
this(client, new ProducerConfigurationData(), schema);
@@ -86,10 +78,7 @@ public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
@Override
public ProducerBuilder<T> clone() {
- ProducerBuilderImpl<T> copy = new ProducerBuilderImpl<>(client, conf.clone(), schema);
- copy.maxPendingMessagesConfigured = maxPendingMessagesConfigured;
- copy.maxPendingMessagesAcrossPartitionsConfigured = maxPendingMessagesAcrossPartitionsConfigured;
- return copy;
+ return new ProducerBuilderImpl<>(client, conf.clone(), schema);
}
@Override
@@ -130,8 +119,7 @@ public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
effectiveInterceptors.add(new org.apache.pulsar.client.impl.tracing.OpenTelemetryProducerInterceptor());
}
- ProducerConfigurationData producerConf = client.applyNoMemoryLimitProducerDefaults(conf,
- maxPendingMessagesConfigured, maxPendingMessagesAcrossPartitionsConfigured);
+ ProducerConfigurationData producerConf = client.applyNoMemoryLimitProducerDefaults(conf);
return effectiveInterceptors == null || effectiveInterceptors.size() == 0
? client.createProducerAsync(producerConf, schema, null)
@@ -140,13 +128,19 @@ public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
@Override
public ProducerBuilder<T> loadConf(Map<String, Object> config) {
+ // 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 by replaying every
+ // property through its setters, so the markers have to be carried over rather than read off
+ // the result.
+ boolean maxPendingMessagesConfigured =
+ conf.isMaxPendingMessagesConfigured() || config.containsKey("maxPendingMessages");
+ boolean maxPendingMessagesAcrossPartitionsConfigured =
+ conf.isMaxPendingMessagesAcrossPartitionsConfigured()
+ || config.containsKey("maxPendingMessagesAcrossPartitions");
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");
+ conf.setMaxPendingMessagesConfigured(maxPendingMessagesConfigured);
+ conf.setMaxPendingMessagesAcrossPartitionsConfigured(maxPendingMessagesAcrossPartitionsConfigured);
return this;
}
@@ -172,7 +166,6 @@ public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
@Override
public ProducerBuilder<T> maxPendingMessages(int maxPendingMessages) {
conf.setMaxPendingMessages(maxPendingMessages);
- maxPendingMessagesConfigured = true;
return this;
}
@@ -180,7 +173,6 @@ public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
@Override
public ProducerBuilder<T> 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 a01c07926844..f89972edd0bb 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
@@ -431,28 +431,29 @@ public class PulsarClientImpl implements PulsarClient {
*
* <p>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.
+ * limit that was never configured is filled in, which is why this reads the markers the
+ * configuration carries rather than inferring it from the values: {@code 0} is both the unset value
+ * and a meaningful explicit one.
*
- * <p>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.
+ * <p>What is filled in here stays marked as unconfigured, so on a partitioned topic
+ * {@link PartitionedProducerImpl} divides only a budget the application actually asked for. A
+ * filled-in budget never lowers a per-producer limit that was asked for.
*
- * <p>Called by {@link ProducerBuilderImpl}, which is what knows whether a limit was configured.
+ * <p>Called by {@link ProducerBuilderImpl}.
*
- * @param conf the requested producer configuration
- * @param maxPendingMessagesConfigured whether the application configured {@code maxPendingMessages}
- * @param maxPendingMessagesAcrossPartitionsConfigured whether the application configured
- * {@code maxPendingMessagesAcrossPartitions}
+ * @param conf the requested producer configuration, carrying the markers that say which limits the
+ * application configured
* @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) {
+ public ProducerConfigurationData applyNoMemoryLimitProducerDefaults(ProducerConfigurationData conf) {
// 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;
+ // configuration was populated. The markers only tell an explicit 0 apart from an unset one.
+ boolean maxPendingMessagesConfigured =
+ conf.isMaxPendingMessagesConfigured() || conf.getMaxPendingMessages() > 0;
+ boolean maxPendingMessagesAcrossPartitionsConfigured =
+ conf.isMaxPendingMessagesAcrossPartitionsConfigured()
+ || conf.getMaxPendingMessagesAcrossPartitions() > 0;
if ((maxPendingMessagesConfigured && maxPendingMessagesAcrossPartitionsConfigured)
|| memoryLimitController.isMemoryLimited()) {
return conf;
@@ -464,10 +465,8 @@ public class PulsarClientImpl implements PulsarClient {
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.
+ // The application asked for no per-producer limit at all, so there is no queue for a
+ // budget to bound. Leaving it unset keeps the resolved configuration honest about that.
maxPendingMessagesAcrossPartitions = 0;
} else {
maxPendingMessagesAcrossPartitions =
@@ -485,6 +484,12 @@ public class PulsarClientImpl implements PulsarClient {
ProducerConfigurationData resolved = conf.clone();
resolved.setMaxPendingMessages(maxPendingMessages);
resolved.setMaxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions);
+ // The setters mark whatever they are given as configured, so restore the markers: what is
+ // filled in here is a default, and a partitioned producer still has to be able to tell it apart
+ // from a limit the application asked for, so that a budget it did ask for is the one that gets
+ // divided.
+ resolved.setMaxPendingMessagesConfigured(maxPendingMessagesConfigured);
+ resolved.setMaxPendingMessagesAcrossPartitionsConfigured(maxPendingMessagesAcrossPartitionsConfigured);
if (log.isDebugEnabled()) {
log.debug("[{}] Client memory limit is disabled, applying default producer pending message limits."
+ " maxPendingMessages: {}, 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 4e6fb77ca796..cc4666ee750d 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
@@ -109,6 +109,30 @@ public class ProducerConfigurationData implements Serializable, Cloneable {
)
private int maxPendingMessagesAcrossPartitions = DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS;
+ /**
+ * Whether the application configured {@link #maxPendingMessages}, and whether it configured
+ * {@link #maxPendingMessagesAcrossPartitions}.
+ *
+ * <p>The unset value of both limits is {@code 0}, which is also a meaningful explicit value ("no
+ * message-count limit" and "no across-partitions budget"), so the value alone cannot tell the two
+ * apart. Recording it here rather than on the builder lets everything that resolves these limits
+ * read the same answer, including {@code PartitionedProducerImpl}, which only sees the
+ * configuration.
+ *
+ * <p>Setting either limit through its setter marks it as configured, so a configuration populated
+ * directly rather than through {@code ProducerBuilderImpl} behaves the same way.
+ *
+ * <p>Deliberately not part of the serialized configuration: {@code loadConf} rebuilds the instance
+ * by replaying every property through its setters, which would mark both limits as configured
+ * whatever the application passed, so {@code ProducerBuilderImpl} restores them across that
+ * round-trip. {@code PulsarClientImpl} likewise restores them after filling in a default, which is
+ * not application input.
+ */
+ @JsonIgnore
+ private boolean maxPendingMessagesConfigured;
+ @JsonIgnore
+ private boolean maxPendingMessagesAcrossPartitionsConfigured;
+
@ApiModelProperty(
name = "messageRoutingMode",
value = "Message routing logic for producers on [partitioned topics]"
@@ -250,6 +274,7 @@ public class ProducerConfigurationData implements Serializable, Cloneable {
public void setMaxPendingMessages(int maxPendingMessages) {
checkArgument(maxPendingMessages >= 0, "maxPendingMessages needs to be >= 0");
this.maxPendingMessages = maxPendingMessages;
+ this.maxPendingMessagesConfigured = true;
}
/**
@@ -264,6 +289,7 @@ public class ProducerConfigurationData implements Serializable, Cloneable {
checkArgument(maxPendingMessagesAcrossPartitions >= 0,
"maxPendingMessagesAcrossPartitions needs to be >= 0");
this.maxPendingMessagesAcrossPartitions = maxPendingMessagesAcrossPartitions;
+ this.maxPendingMessagesAcrossPartitionsConfigured = true;
}
public void setBatchingMaxMessages(int batchingMaxMessages) {
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java
index ce2c200344d9..c7de9c7866a5 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java
@@ -300,14 +300,21 @@ public class PartitionedProducerImplTest {
clientImpl, topicName, producerConfData, 1, null, null, null);
assertEquals(partitionedProducerImpl.getConfiguration().getMaxPendingMessages(), 10);
- // Test set MaxPendingMessagesAcrossPartitions=5
- producerConfData.setMaxPendingMessages(ProducerConfigurationData.DEFAULT_MAX_PENDING_MESSAGES);
+ // Test set MaxPendingMessagesAcrossPartitions=5 with maxPendingMessages left unset. A fresh
+ // configuration is required to express "unset": setting maxPendingMessages back to 0 would mean
+ // "no message-count limit", which is a limit of its own and would win over the budget's share.
+ producerConfData = new ProducerConfigurationData();
+ producerConfData.setMessageRoutingMode(MessageRoutingMode.CustomPartition);
+ producerConfData.setCustomMessageRouter(new CustomMessageRouter());
producerConfData.setMaxPendingMessagesAcrossPartitions(5);
partitionedProducerImpl = new PartitionedProducerImpl(
clientImpl, topicName, producerConfData, 1, null, null, null);
assertEquals(partitionedProducerImpl.getConfiguration().getMaxPendingMessages(), 5);
// Test set maxPendingMessage=10 and MaxPendingMessagesAcrossPartitions=10 with 2 partitions
+ producerConfData = new ProducerConfigurationData();
+ producerConfData.setMessageRoutingMode(MessageRoutingMode.CustomPartition);
+ producerConfData.setCustomMessageRouter(new CustomMessageRouter());
producerConfData.setMaxPendingMessages(10);
producerConfData.setMaxPendingMessagesAcrossPartitions(10);
partitionedProducerImpl = new PartitionedProducerImpl(
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 4b6f9a197d05..5f2506b0835a 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,7 +19,6 @@
package org.apache.pulsar.client.impl;
import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.anyBoolean;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -67,8 +66,8 @@ public class ProducerBuilderImplTest {
// 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.applyNoMemoryLimitProducerDefaults(any(ProducerConfigurationData.class)))
+ .thenAnswer(invocation -> invocation.getArgument(0));
when(client.createProducerAsync(
any(ProducerConfigurationData.class), any(Schema.class), eq(null)))
@@ -139,8 +138,8 @@ public class ProducerBuilderImplTest {
producerBuilderImpl.loadConf(Map.of("maxPendingMessages", 5000));
assertEquals(producerBuilderImpl.getConf().getMaxPendingMessages(), 5000);
- assertTrue(producerBuilderImpl.isMaxPendingMessagesConfigured());
- assertFalse(producerBuilderImpl.isMaxPendingMessagesAcrossPartitionsConfigured());
+ assertTrue(producerBuilderImpl.getConf().isMaxPendingMessagesConfigured());
+ assertFalse(producerBuilderImpl.getConf().isMaxPendingMessagesAcrossPartitionsConfigured());
}
@Test
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 95db3cc61c8f..acc8c6feb348 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,7 +20,6 @@ package org.apache.pulsar.functions.instance;
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;
@@ -113,8 +112,8 @@ public class ContextImplTest {
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.applyNoMemoryLimitProducerDefaults(any(ProducerConfigurationData.class)))
+ .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()));
diff --git a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceProducerTest.java b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceProducerTest.java
index 277f72bcafc0..6bd9ec532134 100644
--- a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceProducerTest.java
+++ b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceProducerTest.java
@@ -260,8 +260,8 @@ public class PerformanceProducerTest extends MockedPulsarServiceBaseTest {
Assert.assertNull(producer.maxOutstanding);
Assert.assertNull(producer.maxPendingMessagesAcrossPartitions);
// pulsar-perf must not configure either limit, so that they stay the client's to decide.
- Assert.assertFalse(builder.isMaxPendingMessagesConfigured());
- Assert.assertFalse(builder.isMaxPendingMessagesAcrossPartitionsConfigured());
+ Assert.assertFalse(builder.getConf().isMaxPendingMessagesConfigured());
+ Assert.assertFalse(builder.getConf().isMaxPendingMessagesAcrossPartitionsConfigured());
// The client resolves its no-memory-limit defaults when the producer is created rather than on
// the builder, so the effective configuration is where they have to show up.