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/10/23 10:08:52 UTC
(pulsar) 21/23: [fix][client] Close the producer when its encryption key cannot be loaded (#27024)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ce0969cf16ca8615662633cb74f57bcedccb0fc9
Author: SeungTaek Lim <na...@gmail.com>
AuthorDate: Fri Oct 23 14:10:36 2026 +0900
[fix][client] Close the producer when its encryption key cannot be loaded (#27024)
(cherry picked from commit c2b5d91c6f413d4f2d5cd9dfd707e61d7879e24b)
---
.../impl/ProducerEncryptionKeyFailureTest.java | 83 +++++++++++++++
.../apache/pulsar/client/impl/ProducerBase.java | 7 ++
.../apache/pulsar/client/impl/ProducerImpl.java | 57 ++++++----
.../pulsar/client/impl/PulsarClientImpl.java | 6 ++
.../pulsar/client/impl/ProducerImplTest.java | 75 +++++++++++++
.../pulsar/client/impl/PulsarClientImplTest.java | 117 +++++++++++++++++++++
6 files changed, 327 insertions(+), 18 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerEncryptionKeyFailureTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerEncryptionKeyFailureTest.java
new file mode 100644
index 000000000000..379243efc66b
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerEncryptionKeyFailureTest.java
@@ -0,0 +1,83 @@
+/*
+ * 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.client.impl;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.EncryptionKeyInfo;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.awaitility.Awaitility;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-impl")
+public class ProducerEncryptionKeyFailureTest extends ProducerConsumerBase {
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ /** A key reader that cannot find any key, as a misconfigured deployment would. */
+ private static final class MissingKeyReader implements CryptoKeyReader {
+ @Override
+ public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> metadata) {
+ return null;
+ }
+
+ @Override
+ public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> metadata) {
+ return null;
+ }
+ }
+
+ @Test(timeOut = 60_000)
+ public void testProducerWhoseKeyCannotBeLoadedIsNotLeftBehind() throws Exception {
+ String topic = "persistent://my-property/my-ns/crypto-key-failure-" + System.nanoTime();
+ // Created up front so that the broker can report its publishers whether or not a producer connects.
+ admin.topics().createNonPartitionedTopic(topic);
+ PulsarClientImpl client = (PulsarClientImpl) pulsarClient;
+
+ expectThrows(PulsarClientException.CryptoException.class, () -> pulsarClient.newProducer()
+ .topic(topic)
+ .addEncryptionKey("missing-key")
+ .cryptoKeyReader(new MissingKeyReader())
+ .create());
+
+ // The connection is opened by another thread: give it time to register the producer, which it must not.
+ Awaitility.await().during(3, TimeUnit.SECONDS).atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
+ assertTrue(admin.topics().getStats(topic).getPublishers().isEmpty(),
+ "A producer whose creation failed must not stay registered on the broker"));
+ assertEquals(client.producersCount(), 0, "A producer whose creation failed must not stay in the client");
+ }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBase.java
index 7cfcec64e896..1b5bef3f9c63 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBase.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBase.java
@@ -147,6 +147,13 @@ public abstract class ProducerBase<T> extends HandlerState implements Producer<T
@Override
public abstract CompletableFuture<Void> closeAsync();
+ /** Whether the producer has reached a state it does not leave: closing, closed, failed, terminated or fenced. */
+ protected boolean isTerminalState() {
+ State state = getState();
+ return state == State.Closing || state == State.Closed || state == State.Failed
+ || state == State.Terminated || state == State.ProducerFenced;
+ }
+
@Override
public String getTopic() {
return topic;
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 393b685e8525..16a220337c87 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
@@ -277,24 +277,6 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
this.msgCrypto = null;
}
- if (this.msgCrypto != null) {
- // Regenerate data key cipher at fixed interval
- keyGeneratorTask = client.eventLoopGroup().scheduleWithFixedDelay(catchingAndLoggingThrowables(() -> {
- try {
- msgCrypto.addPublicKeyCipher(conf.getEncryptionKeys(), conf.getCryptoKeyReader());
- } catch (CryptoException e) {
- if (!producerCreatedFuture.isDone()) {
- log.warn().log("Failed to add public key cipher");
- producerCreatedFuture.completeExceptionally(
- PulsarClientException.wrap(e,
- String.format("The producer %s of the topic %s "
- + "adds the public key cipher was failed",
- producerName, topic)));
- }
- }
- }), 0L, 4L, TimeUnit.HOURS);
- }
-
if (conf.getSendTimeoutMs() > 0) {
sendTimeout = client.timer().newTimeout(this, conf.getSendTimeoutMs(), TimeUnit.MILLISECONDS);
}
@@ -345,6 +327,9 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
this.connectionHandler = initConnectionHandler();
setChunkMaxMessageSize();
+ if (this.msgCrypto != null) {
+ scheduleKeyGeneratorTask();
+ }
grabCnx();
producersOpenedCounter.increment();
}
@@ -3208,6 +3193,42 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
this.connectionHandler.setClientCnx(clientCnx);
}
+ /**
+ * Loads the public key cipher now and refreshes it at a fixed interval. Called once the producer is initialized
+ * and before it connects: a key that cannot be loaded fails the creation and closes the producer, which uses the
+ * counters, the stats and the connection handler, and a connection failure that ends the creation
+ * ({@link #connectionFailed}) must find the handle of the task to cancel it. {@link #closeAsync()} is
+ * synchronized, so the task cannot close the producer before the handle is stored here and the close can cancel
+ * it; when the producer has reached a terminal state by then, from the scheduling thread for instance, the handle
+ * is cancelled here instead.
+ */
+ private synchronized void scheduleKeyGeneratorTask() {
+ keyGeneratorTask = client.eventLoopGroup().scheduleWithFixedDelay(
+ catchingAndLoggingThrowables(this::addPublicKeyCipher), 0L, 4L, TimeUnit.HOURS);
+ if (isTerminalState()) {
+ keyGeneratorTask.cancel(false);
+ }
+ }
+
+ private void addPublicKeyCipher() {
+ try {
+ msgCrypto.addPublicKeyCipher(conf.getEncryptionKeys(), conf.getCryptoKeyReader());
+ } catch (CryptoException e) {
+ if (!producerCreatedFuture.isDone()) {
+ log.warn().log("Failed to add public key cipher");
+ producerCreatedFuture.completeExceptionally(
+ PulsarClientException.wrap(e,
+ String.format("The producer %s of the topic %s "
+ + "adds the public key cipher was failed",
+ producerName, topic)));
+ // The creation has failed, so close the producer as the other creation failures do. Otherwise it
+ // stays in the client and keeps the connection it is about to open, which registers it on the
+ // broker and then reconnects it as if it had been created.
+ closeAsync();
+ }
+ }
+ }
+
void grabCnx() {
this.connectionHandler.grabCnx();
}
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 5b88d3cf8761..a37270342347 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
@@ -1013,6 +1013,12 @@ public class PulsarClientImpl implements PulsarClient {
Optional.empty());
}
producers.add(producer);
+ // The creation can fail before this line runs, from another thread or from a connection that fails at
+ // once, in which case the producer has already closed or failed and removed itself from the client. Do
+ // not keep it then.
+ if (producer.isTerminalState()) {
+ cleanupProducer(producer);
+ }
}).exceptionally(ex -> {
log.warn().attr("topic", topic).exceptionMessage(ex).log("Failed to get partitioned topic metadata");
producerCreatedFuture.completeExceptionally(ex);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
index a881701e6bcc..91f938289488 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
@@ -24,6 +24,7 @@ import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
@@ -66,6 +67,7 @@ import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.client.api.CryptoKeyReader;
import org.apache.pulsar.client.api.MessageCrypto;
+import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
@@ -79,6 +81,7 @@ import org.apache.pulsar.client.impl.metrics.LatencyHistogram;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.protocol.ByteBufPair;
import org.apache.pulsar.common.protocol.Commands;
+import org.apache.pulsar.common.util.FutureUtil;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.testng.annotations.DataProvider;
@@ -327,6 +330,78 @@ public class ProducerImplTest {
return conf;
}
+ @DataProvider
+ public Object[][] keyTaskTiming() {
+ return new Object[][]{{true}, {false}};
+ }
+
+ /**
+ * A key that cannot be loaded fails the creation, and the producer is closed and taken out of the client with
+ * its key task cancelled, whether the task runs while the constructor is still scheduling it ({@code inline})
+ * or later.
+ */
+ @Test(dataProvider = "keyTaskTiming")
+ @SuppressWarnings("unchecked")
+ public void testKeyThatCannotBeLoadedClosesTheProducer(boolean inline) {
+ PulsarClientImpl client = mockedPulsarClient();
+ EventLoopGroup eventLoopGroup = client.eventLoopGroup();
+ ScheduledFuture<?> keyTask = mock(ScheduledFuture.class);
+ AtomicReference<Runnable> scheduled = new AtomicReference<>();
+ doAnswer(invocation -> {
+ Runnable task = invocation.getArgument(0);
+ if (inline) {
+ task.run();
+ } else {
+ scheduled.set(task);
+ }
+ return keyTask;
+ }).when(eventLoopGroup).scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any());
+ ProducerConfigurationData conf = new ProducerConfigurationData();
+ conf.setEncryptionKeys(new TreeSet<>(Collections.singleton("missing-key")));
+ conf.setCryptoKeyReader(mock(CryptoKeyReader.class)); // returns no key
+ CompletableFuture<Producer<byte[]>> created = new CompletableFuture<>();
+
+ ProducerImpl<byte[]> producer = new ProducerImpl<>(client, "persistent://public/default/key-failure", conf,
+ created, 0, Schema.BYTES, null, Optional.empty());
+ if (!inline) {
+ scheduled.get().run();
+ }
+
+ assertTrue(created.isCompletedExceptionally(), "The creation must fail");
+ assertThatThrownBy(created::join).hasCauseInstanceOf(PulsarClientException.CryptoException.class);
+ assertEquals(producer.getState(), HandlerState.State.Closed);
+ verify(keyTask).cancel(false);
+ verify(client).cleanupProducer(producer);
+ }
+
+ /**
+ * A connection failure that ends the creation runs the terminal cleanup; the key task scheduled by the
+ * constructor must be cancelled by it, so it is scheduled before the producer connects.
+ */
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testKeyTaskIsCancelledWhenTheConnectionFailsTheCreation() {
+ PulsarClientImpl client = mockedPulsarClient();
+ EventLoopGroup eventLoopGroup = client.eventLoopGroup();
+ ScheduledFuture<?> keyTask = mock(ScheduledFuture.class);
+ doReturn(keyTask).when(eventLoopGroup).scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any());
+ when(client.getConnection(anyString(), anyInt())).thenReturn(
+ FutureUtil.failedFuture(new PulsarClientException.AuthorizationException("not allowed")));
+ ProducerConfigurationData conf = new ProducerConfigurationData();
+ conf.setEncryptionKeys(new TreeSet<>(Collections.singleton("key")));
+ conf.setCryptoKeyReader(mock(CryptoKeyReader.class));
+ CompletableFuture<Producer<byte[]>> created = new CompletableFuture<>();
+
+ ProducerImpl<byte[]> producer = new ProducerImpl<>(client, "persistent://public/default/connection-failure",
+ conf, created, 0, Schema.BYTES, null, Optional.empty());
+
+ assertTrue(created.isCompletedExceptionally(), "The creation must fail");
+ assertThatThrownBy(created::join).hasCauseInstanceOf(PulsarClientException.AuthorizationException.class);
+ assertEquals(producer.getState(), HandlerState.State.Failed);
+ verify(keyTask).cancel(false);
+ verify(client).cleanupProducer(producer);
+ }
+
/** A client mock whose stubs satisfy the {@link ProducerImpl} constructor. */
private static PulsarClientImpl mockedPulsarClient() {
PulsarClientImpl client = mock(PulsarClientImpl.class);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java
index 8e01cc08c333..54050cb8eda5 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java
@@ -54,17 +54,23 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
import java.util.regex.Pattern;
import lombok.Cleanup;
+import org.apache.pulsar.client.api.MessageRoutingMode;
+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.client.api.PulsarClientSharedResources;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.ServiceUrlProvider;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.client.util.ScheduledExecutorProvider;
@@ -76,6 +82,7 @@ import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.util.netty.EventLoopUtil;
import org.mockito.Mockito;
import org.testng.annotations.AfterMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
/**
@@ -108,6 +115,116 @@ public class PulsarClientImplTest {
assertTrue(client.isClosed());
}
+ @DataProvider
+ public Object[][] partitions() {
+ return new Object[][]{{0}, {2}};
+ }
+
+ /**
+ * Cancelling the creation future does not stop the creation: the producer still comes up, and the client keeps
+ * owning it, so it is closed with the client.
+ */
+ @Test(dataProvider = "partitions")
+ @SuppressWarnings("unchecked")
+ public void testCancelledCreationLeavesTheProducerOwnedByTheClient(int partitions) throws Exception {
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setServiceUrl("pulsar://localhost:6650");
+ initializeEventLoopGroup(conf);
+ LookupService lookup = mock(LookupService.class);
+ CompletableFuture<PartitionedTopicMetadata> metadata = new CompletableFuture<>();
+ when(lookup.getPartitionedTopicMetadata(any(TopicName.class), anyBoolean(), anyBoolean()))
+ .thenReturn(metadata);
+ when(lookup.getBroker(any()))
+ .thenReturn(CompletableFuture.completedFuture(new LookupTopicResult(
+ mock(InetSocketAddress.class), mock(InetSocketAddress.class), false)));
+ ConnectionPool pool = mock(ConnectionPool.class);
+ ClientCnx cnx = mock(ClientCnx.class);
+ ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+ Channel channel = mock(Channel.class);
+ when(channel.remoteAddress()).thenReturn(mock(SocketAddress.class));
+ when(channel.eventLoop()).thenReturn(eventLoopGroup.next());
+ when(ctx.channel()).thenReturn(channel);
+ when(ctx.writeAndFlush(any(), any(ChannelPromise.class))).thenReturn(mock(ChannelFuture.class));
+ when(ctx.voidPromise()).thenReturn(mock(ChannelPromise.class));
+ when(cnx.channel()).thenReturn(channel);
+ when(cnx.ctx()).thenReturn(ctx);
+ ProducerResponse response = mock(ProducerResponse.class);
+ when(response.getProducerName()).thenReturn("cancelled-creation-producer");
+ when(cnx.sendRequestWithId(any(ByteBuf.class), anyLong()))
+ .thenReturn(CompletableFuture.completedFuture(response));
+ when(pool.getConnection(any(InetSocketAddress.class), any(InetSocketAddress.class), anyInt()))
+ .thenReturn(CompletableFuture.completedFuture(cnx));
+ PulsarClientImpl client = new PulsarClientImpl(conf, eventLoopGroup, pool);
+ client.setLookup(lookup);
+ try {
+ ProducerConfigurationData producerConf = new ProducerConfigurationData();
+ producerConf.setTopicName("persistent://public/default/cancelled-creation");
+ producerConf.setMessageRoutingMode(MessageRoutingMode.RoundRobinPartition);
+ CompletableFuture<Producer<byte[]>> creation = client.createProducerAsync(producerConf, Schema.BYTES);
+
+ // Cancelled before the metadata is known, which is before the producer exists.
+ creation.cancel(true);
+ metadata.complete(new PartitionedTopicMetadata(partitions));
+
+ assertEquals(client.producersCount(), 1, "The client must keep owning the producer");
+ } finally {
+ client.close();
+ }
+ assertEquals(client.producersCount(), 0, "Closing the client must close the producer it kept");
+ }
+
+ @DataProvider
+ public Object[][] endedBeforeRegistration() {
+ return new Object[][]{
+ {"closed", (Consumer<ProducerImpl<?>>) ProducerImpl::closeAsync},
+ {"failed", (Consumer<ProducerImpl<?>>) producer ->
+ producer.connectionFailed(new PulsarClientException.AuthorizationException("not allowed"))},
+ };
+ }
+
+ /**
+ * A producer whose creation fails before the client has registered it, from another thread or from a
+ * connection that fails at once, removes itself from the client while it is not there yet. The client must not
+ * keep it afterwards.
+ */
+ @Test(dataProvider = "endedBeforeRegistration")
+ @SuppressWarnings("unchecked")
+ public void testProducerEndedBeforeItIsRegisteredIsNotKept(String how, Consumer<ProducerImpl<?>> end)
+ throws Exception {
+ LookupService lookup = mock(LookupService.class);
+ when(lookup.getPartitionedTopicMetadata(any(TopicName.class), anyBoolean(), anyBoolean()))
+ .thenReturn(CompletableFuture.completedFuture(new PartitionedTopicMetadata()));
+ when(lookup.getBroker(any())).thenReturn(new CompletableFuture<>());
+ ConnectionPool pool = mock(ConnectionPool.class);
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setServiceUrl("pulsar://localhost:6650");
+ PulsarClientImpl client = new PulsarClientImpl(conf, eventLoopGroup, pool) {
+ @Override
+ protected <T> ProducerImpl<T> newProducerImpl(String topic, int partitionIndex,
+ ProducerConfigurationData producerConf, Schema<T> schema,
+ ProducerInterceptors interceptors,
+ CompletableFuture<Producer<T>> producerCreatedFuture,
+ Optional<String> overrideProducerName) {
+ ProducerImpl<T> producer = super.newProducerImpl(topic, partitionIndex, producerConf, schema,
+ interceptors, producerCreatedFuture, overrideProducerName);
+ // Before createProducerAsync() has added the producer.
+ end.accept(producer);
+ return producer;
+ }
+ };
+ client.setLookup(lookup);
+ try {
+ ProducerConfigurationData producerConf = new ProducerConfigurationData();
+ producerConf.setTopicName("persistent://public/default/closed-before-registered");
+ client.createProducerAsync(producerConf, Schema.BYTES);
+
+ assertEquals(client.producersCount(), 0,
+ "A producer " + how + " before it was registered must not be kept");
+ } finally {
+ client.close();
+ }
+ }
+
@Test
@SuppressWarnings("unchecked")
public void testConsumerIsClosed() throws Exception {