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 cc4b53d101705..8a2022678b838 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 @@ -366,4 +366,107 @@ public void testMemoryLimitedClientKeepsUnboundedPendingMessages() throws Except 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 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 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 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 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 a0f2105bc1997..3d8e7cb204561 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.github.merlimat.slog.Logger; import io.netty.util.Timeout; @@ -88,14 +86,23 @@ public PartitionedProducerImpl(PulsarClientImpl client, String topic, ProducerCo ? 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 11915140ceab9..9589d31a1470c 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 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); @@ -86,10 +78,7 @@ public ProducerBuilder schema(Schema schema) { @Override public ProducerBuilder clone() { - ProducerBuilderImpl copy = new ProducerBuilderImpl<>(client, conf.clone(), schema); - copy.maxPendingMessagesConfigured = maxPendingMessagesConfigured; - copy.maxPendingMessagesAcrossPartitionsConfigured = maxPendingMessagesAcrossPartitionsConfigured; - return copy; + return new ProducerBuilderImpl<>(client, conf.clone(), schema); } @Override @@ -131,8 +120,7 @@ public CompletableFuture> createAsync() { client.instrumentProvider())); } - ProducerConfigurationData producerConf = client.applyNoMemoryLimitProducerDefaults(conf, - maxPendingMessagesConfigured, maxPendingMessagesAcrossPartitionsConfigured); + ProducerConfigurationData producerConf = client.applyNoMemoryLimitProducerDefaults(conf); return effectiveInterceptors == null || effectiveInterceptors.size() == 0 ? client.createProducerAsync(producerConf, schema, null) @@ -141,13 +129,19 @@ public CompletableFuture> createAsync() { @Override public ProducerBuilder loadConf(Map 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; } @@ -173,7 +167,6 @@ public ProducerBuilder sendTimeout(int sendTimeout, @NonNull TimeUnit unit) { @Override public ProducerBuilder maxPendingMessages(int maxPendingMessages) { conf.setMaxPendingMessages(maxPendingMessages); - maxPendingMessagesConfigured = true; return this; } @@ -181,7 +174,6 @@ 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 221c75436c063..52c9de617e474 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 @@ -716,31 +716,32 @@ public CompletableFuture> createProducerAsync(ProducerConfigurat * *

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. * - *

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. + *

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. * - *

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. + *

Called by {@link ProducerBuilderImpl}. 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} + * @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; @@ -752,10 +753,8 @@ public ProducerConfigurationData applyNoMemoryLimitProducerDefaults(ProducerConf 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 = @@ -773,6 +772,12 @@ public ProducerConfigurationData applyNoMemoryLimitProducerDefaults(ProducerConf 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); log.debug().attr("topic", conf.getTopicName()) .attr("maxPendingMessages", maxPendingMessages) .attr("maxPendingMessagesAcrossPartitions", 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 02c25bece9053..0f1861e30f5c3 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}. + * + *

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. + * + *

Setting either limit through its setter marks it as configured, so a configuration populated + * directly rather than through {@code ProducerBuilderImpl} behaves the same way. + * + *

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; + @Schema( name = "messageRoutingMode", description = "Message routing logic for producers on [partitioned topics]" @@ -250,6 +274,7 @@ public void setProducerName(String producerName) { public void setMaxPendingMessages(int maxPendingMessages) { checkArgument(maxPendingMessages >= 0, "maxPendingMessages needs to be >= 0"); this.maxPendingMessages = maxPendingMessages; + this.maxPendingMessagesConfigured = true; } /** @@ -264,6 +289,7 @@ public void setMaxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPa 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 f2acdfa3f17b7..3c96e7aa8fb2c 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 @@ -309,14 +309,21 @@ public void testMaxPendingQueueSize() throws Exception { 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 bdae20c1ed22e..ab727ce1b3e69 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.doReturn; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; @@ -68,8 +67,8 @@ public void setup() { // 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)); doReturn(CompletableFuture.completedFuture(producer)) .when(client).createProducerAsync( @@ -140,8 +139,8 @@ public void testLoadConfWithAPositiveMaxPendingMessages() { 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 d7fd8ad19979a..96e2dc5d75ad7 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 @@ 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; @@ -114,8 +113,8 @@ public void setup() throws PulsarClientException { 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()));