You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/10/22 20:28:36 UTC
(pulsar) branch master updated: [fix][client] Client interceptors close method not called after producer/consumer/reader close (#23830)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new a883ea4b1d7d [fix][client] Client interceptors close method not called after producer/consumer/reader close (#23830)
a883ea4b1d7d is described below
commit a883ea4b1d7df9b10d7806189a0acb203126e7c8
Author: Yike Xiao <km...@live.com>
AuthorDate: Fri Oct 23 04:28:28 2026 +0800
[fix][client] Client interceptors close method not called after producer/consumer/reader close (#23830)
Co-authored-by: Matteo Merli <mm...@apache.org>
---
.../apache/pulsar/client/api/InterceptorsTest.java | 88 ++++++++++++++++++++++
.../pulsar/client/api/ConsumerInterceptor.java | 3 +
.../pulsar/client/api/ProducerInterceptor.java | 3 +
.../pulsar/client/api/ReaderInterceptor.java | 3 +
.../api/interceptor/ProducerInterceptor.java | 3 +
.../apache/pulsar/client/impl/ConsumerBase.java | 10 +++
.../apache/pulsar/client/impl/ConsumerImpl.java | 6 +-
.../pulsar/client/impl/ConsumerInterceptors.java | 3 +-
.../client/impl/MultiTopicsConsumerImpl.java | 5 +-
.../client/impl/PartitionedProducerImpl.java | 32 +++++++-
.../apache/pulsar/client/impl/ProducerBase.java | 11 +++
.../apache/pulsar/client/impl/ProducerImpl.java | 3 +
.../pulsar/client/impl/ProducerInterceptors.java | 3 +-
.../pulsar/client/impl/ConsumerImplTest.java | 11 +++
.../pulsar/client/impl/ProducerImplTest.java | 24 +++++-
15 files changed, 197 insertions(+), 11 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java
index 622542a621ca..b72016fddce9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java
@@ -18,6 +18,12 @@
*/
package org.apache.pulsar.client.api;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -1113,6 +1119,88 @@ public class InterceptorsTest extends SharedPulsarBaseTest {
consumer.close();
}
+ @Test(dataProvider = "topicPartition")
+ public void testProducerInterceptorClosedOnceOnProducerClose(int topicPartition) throws Exception {
+ String topic = newTopicName();
+ if (topicPartition > 0) {
+ admin.topics().createPartitionedTopic(topic, topicPartition);
+ }
+ org.apache.pulsar.client.api.interceptor.ProducerInterceptor interceptor =
+ mock(org.apache.pulsar.client.api.interceptor.ProducerInterceptor.class);
+ when(interceptor.eligible(any())).thenReturn(true);
+ when(interceptor.beforeSend(any(), any())).thenAnswer(invocation -> invocation.getArgument(1));
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topic)
+ .intercept(interceptor)
+ .create();
+ producer.send("msg");
+ verify(interceptor).beforeSend(any(), any());
+ verify(interceptor, never()).close();
+
+ producer.close();
+ producer.close();
+
+ verify(interceptor).close();
+ }
+
+ @Test
+ public void testProducerInterceptorSharedByTwoProducersClosedForEach() throws Exception {
+ String topic = newTopicName();
+ org.apache.pulsar.client.api.interceptor.ProducerInterceptor interceptor =
+ mock(org.apache.pulsar.client.api.interceptor.ProducerInterceptor.class);
+ ProducerBuilder<String> builder = pulsarClient.newProducer(Schema.STRING)
+ .topic(topic)
+ .intercept(interceptor);
+ Producer<String> producer1 = builder.create();
+ Producer<String> producer2 = builder.create();
+
+ producer1.close();
+ verify(interceptor).close();
+
+ producer2.close();
+ verify(interceptor, times(2)).close();
+ }
+
+ @Test(dataProvider = "topicPartition")
+ public void testConsumerInterceptorClosedOnceOnConsumerClose(int topicPartition) throws Exception {
+ String topic = newTopicName();
+ if (topicPartition > 0) {
+ admin.topics().createPartitionedTopic(topic, topicPartition);
+ }
+ ConsumerInterceptor<String> interceptor = mock(ConsumerInterceptor.class);
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("my-subscription")
+ .intercept(interceptor)
+ .subscribe();
+ verify(interceptor, never()).close();
+
+ consumer.close();
+ consumer.close();
+
+ verify(interceptor).close();
+ }
+
+ @Test(dataProvider = "topicPartition")
+ public void testReaderInterceptorClosedOnceOnReaderClose(int topicPartition) throws Exception {
+ String topic = newTopicName();
+ if (topicPartition > 0) {
+ admin.topics().createPartitionedTopic(topic, topicPartition);
+ }
+ ReaderInterceptor<String> interceptor = mock(ReaderInterceptor.class);
+ Reader<String> reader = pulsarClient.newReader(Schema.STRING)
+ .topic(topic)
+ .startMessageId(MessageId.earliest)
+ .intercept(interceptor)
+ .create();
+ verify(interceptor, never()).close();
+
+ reader.close();
+ reader.close();
+
+ verify(interceptor).close();
+ }
+
private void produceAndConsume(int msgCount, Producer<byte[]> producer, Reader<byte[]> reader)
throws PulsarClientException {
for (int i = 0; i < msgCount; i++) {
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerInterceptor.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerInterceptor.java
index 1beea3adba23..c17e99ac35d0 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerInterceptor.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerInterceptor.java
@@ -38,6 +38,9 @@ public interface ConsumerInterceptor<T> extends AutoCloseable {
/**
* Close the interceptor.
+ *
+ * <p>This method is called once when the consumer that this interceptor is attached to is closed. An interceptor
+ * instance that is attached to several consumers is closed once for each of them.
*/
void close();
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerInterceptor.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerInterceptor.java
index 4989e80cf4b9..ba25f10f9026 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerInterceptor.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerInterceptor.java
@@ -39,6 +39,9 @@ public interface ProducerInterceptor<T> extends AutoCloseable {
/**
* Close the interceptor.
+ *
+ * <p>This method is called once when the producer that this interceptor is attached to is closed. An interceptor
+ * instance that is attached to several producers is closed once for each of them.
*/
void close();
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ReaderInterceptor.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ReaderInterceptor.java
index 3a2e93d9c35a..a56c9d6c56ef 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ReaderInterceptor.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ReaderInterceptor.java
@@ -31,6 +31,9 @@ public interface ReaderInterceptor<T> {
/**
* Close the interceptor.
+ *
+ * <p>This method is called once when the reader that this interceptor is attached to is closed. An interceptor
+ * instance that is attached to several readers is closed once for each of them.
*/
void close();
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/interceptor/ProducerInterceptor.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/interceptor/ProducerInterceptor.java
index 3eb7e7d897d7..44f075a6dc4e 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/interceptor/ProducerInterceptor.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/interceptor/ProducerInterceptor.java
@@ -46,6 +46,9 @@ public interface ProducerInterceptor extends AutoCloseable {
/**
* Close the interceptor.
+ *
+ * <p>This method is called once when the producer that this interceptor is attached to is closed. An interceptor
+ * instance that is attached to several producers is closed once for each of them.
*/
void close();
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
index fbccb61a0db6..b558abf15dd9 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
@@ -110,6 +110,7 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
AtomicIntegerFieldUpdater.newUpdater(ConsumerBase.class, "currentReceiverQueueSize");
protected final Schema<T> schema;
protected final ConsumerInterceptors<T> interceptors;
+ private final AtomicBoolean interceptorsClosed = new AtomicBoolean();
protected final BatchReceivePolicy batchReceivePolicy;
protected final ConcurrentLinkedQueue<OpBatchReceive<T>> pendingBatchReceives;
@SuppressWarnings("rawtypes")
@@ -993,6 +994,15 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
}
}
+ /**
+ * Close the interceptors once this consumer is closed. Later calls do nothing.
+ */
+ protected void closeInterceptors() {
+ if (interceptors != null && interceptorsClosed.compareAndSet(false, true)) {
+ interceptors.close();
+ }
+ }
+
protected boolean canEnqueueMessage(Message<T> message) {
// Default behavior, can be overridden in subclasses
return true;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 19da8d98f1d5..df9f1987d5a1 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -1221,7 +1221,10 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
closeConsumerTasks();
deregisterFromClientCnx();
client.cleanupConsumer(this);
- failPendingReceive().whenComplete((r, t) -> closeFuture.complete(null));
+ failPendingReceive().whenComplete((r, t) -> {
+ closeInterceptors();
+ closeFuture.complete(null);
+ });
return compositeCloseFuture;
}
@@ -1261,6 +1264,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
// fail all pending-receive futures to notify application
failPendingReceive().whenComplete((r, t) -> {
+ closeInterceptors();
if (exception != null) {
closeFuture.completeExceptionally(exception);
} else {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerInterceptors.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerInterceptors.java
index c2dc61e7fe94..c10a03162770 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerInterceptors.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerInterceptors.java
@@ -19,7 +19,6 @@
package org.apache.pulsar.client.impl;
import java.io.Closeable;
-import java.io.IOException;
import java.util.List;
import java.util.Set;
import lombok.CustomLog;
@@ -213,7 +212,7 @@ public class ConsumerInterceptors<T> implements Closeable {
}
@Override
- public void close() throws IOException {
+ public void close() {
for (int i = 0, interceptorsSize = interceptors.size(); i < interceptorsSize; i++) {
try {
interceptors.get(i).close();
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
index 323eeac0839a..95218f6730a1 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -27,7 +27,6 @@ import com.google.common.collect.Lists;
import io.github.merlimat.slog.Logger;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -677,6 +676,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
return failPendingReceive();
}, internalPinnedExecutor)
.whenComplete((r, ex) -> {
+ closeInterceptors();
if (ex == null) {
closeFuture.complete(null);
} else {
@@ -1703,8 +1703,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
}
@Override
- public void close() throws IOException {
- multiTopicInterceptors.close();
+ public void close() {
}
};
}
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 3d8e7cb20456..2e66e691a1f4 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
@@ -63,6 +63,7 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
private final Map<Integer, ProducerImpl<T>> producers = new ConcurrentHashMap<>();
private final MessageRouter routerPolicy;
private final PartitionedTopicProducerStatsRecorderImpl stats;
+ private final ProducerInterceptors internalProducerInterceptors;
private TopicMetadata topicMetadata;
private final int firstPartitionIndex;
private String overrideProducerName;
@@ -82,6 +83,9 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
.build();
this.topicMetadata = new TopicMetadataImpl(numPartitions);
this.routerPolicy = getMessageRouter();
+ this.internalProducerInterceptors = interceptors != null
+ ? createInternalProducerInterceptors(interceptors)
+ : null;
stats = client.getConfiguration().getStatsIntervalSeconds() > 0
? new PartitionedTopicProducerStatsRecorderImpl()
: null;
@@ -233,7 +237,7 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
return producers.computeIfAbsent(partitionIndex, (idx) -> {
String partitionName = TopicName.get(topic).getPartition(idx).toString();
return client.newProducerImpl(partitionName, idx,
- conf, schema, interceptors, new CompletableFuture<>(), overrideProducerName);
+ conf, schema, internalProducerInterceptors, new CompletableFuture<>(), overrideProducerName);
});
}
@@ -353,6 +357,7 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
closeFail.compareAndSet(null, ex);
}
if (completed.decrementAndGet() == 0) {
+ closeInterceptors();
if (closeFail.get() == null) {
setState(State.Closed);
closeFuture.complete(null);
@@ -483,6 +488,31 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
}
}
+ private ProducerInterceptors createInternalProducerInterceptors(ProducerInterceptors interceptors) {
+ return new ProducerInterceptors(Collections.emptyList()) {
+ @Override
+ public Message beforeSend(Producer producer, Message message) {
+ return interceptors.beforeSend(producer, message);
+ }
+
+ @Override
+ public void onSendAcknowledgement(Producer producer, Message message, MessageId msgId,
+ Throwable exception) {
+ interceptors.onSendAcknowledgement(producer, message, msgId, exception);
+ }
+
+ @Override
+ public void onPartitionsChange(String topicName, int partitions) {
+ // do nothing
+ }
+
+ @Override
+ public void close() {
+ // do nothing
+ }
+ };
+ }
+
private TimerTask partitionsAutoUpdateTimerTask = new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
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 b7085d28ee2a..7cfcec64e896 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
@@ -22,6 +22,7 @@ import static com.google.common.base.Preconditions.checkArgument;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
@@ -41,6 +42,7 @@ public abstract class ProducerBase<T> extends HandlerState implements Producer<T
protected final ProducerConfigurationData conf;
protected final Schema<T> schema;
protected final ProducerInterceptors interceptors;
+ private final AtomicBoolean interceptorsClosed = new AtomicBoolean();
protected final Map<SchemaHash, byte[]> schemaCache = new ConcurrentHashMap<>();
protected volatile MultiSchemaMode multiSchemaMode = MultiSchemaMode.Auto;
@@ -178,6 +180,15 @@ public abstract class ProducerBase<T> extends HandlerState implements Producer<T
}
}
+ /**
+ * Close the interceptors once this producer is closed. Later calls do nothing.
+ */
+ protected void closeInterceptors() {
+ if (interceptors != null && interceptorsClosed.compareAndSet(false, true)) {
+ interceptors.close();
+ }
+ }
+
@Override
public String toString() {
return "ProducerBase{" + "topic='" + 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 3bbdee956b47..393b685e8525 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
@@ -1459,6 +1459,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
if (cnx == null || (currentState != State.Ready && currentState != State.RegisteringSchema)) {
log.info("Closed Producer (not connected)");
closeAndClearPendingMessages();
+ closeInterceptors();
return CompletableFuture.completedFuture(null);
}
@@ -1472,8 +1473,10 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
// connection did break in the meantime. In any case, the producer is gone.
log.info("Closed Producer");
closeAndClearPendingMessages();
+ closeInterceptors();
closeFuture.complete(null);
} else {
+ closeInterceptors();
closeFuture.completeExceptionally(exception);
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerInterceptors.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerInterceptors.java
index 9d7519343bb7..c87cadee8dcc 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerInterceptors.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerInterceptors.java
@@ -19,7 +19,6 @@
package org.apache.pulsar.client.impl;
import java.io.Closeable;
-import java.io.IOException;
import java.util.List;
import lombok.CustomLog;
import org.apache.pulsar.client.api.Message;
@@ -114,7 +113,7 @@ public class ProducerInterceptors implements Closeable {
}
@Override
- public void close() throws IOException {
+ public void close() {
for (ProducerInterceptor interceptor : interceptors) {
try {
interceptor.close();
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java
index 8ada68d3ef73..9221add8590c 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java
@@ -607,6 +607,17 @@ public class ConsumerImplTest {
}
}
+ @Test
+ public void testCloseNotConnectedConsumerClosesInterceptorsOnce() throws Exception {
+ ConsumerInterceptor<byte[]> interceptor = mock(ConsumerInterceptor.class);
+ createConsumer(new ConsumerConfigurationData<>(), topic, interceptor);
+
+ consumer.closeAsync().get(5, TimeUnit.SECONDS);
+ consumer.closeAsync().get(5, TimeUnit.SECONDS);
+
+ verify(interceptor).close();
+ }
+
@Test
public void testMessageEnqueuedAfterCloseIsDiscarded() {
consumerConf = new ConsumerConfigurationData<>();
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 02be50b81ccf..a881701e6bcc 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
@@ -69,6 +69,7 @@ import org.apache.pulsar.client.api.MessageCrypto;
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
import org.apache.pulsar.client.impl.ProducerImpl.OpSendMsg;
import org.apache.pulsar.client.impl.ProducerImpl.OpSendMsgQueue;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
@@ -305,6 +306,20 @@ public class ProducerImplTest {
connectionFuture.completeExceptionally(new PulsarClientException("producer closed during connect"));
}
+ @Test
+ public void testCloseNotConnectedProducerClosesInterceptorsOnce() throws Exception {
+ PulsarClientImpl client = mockedPulsarClient();
+ when(client.getConnection(Mockito.anyString(), Mockito.anyInt())).thenReturn(new CompletableFuture<>());
+ ProducerInterceptor interceptor = mock(ProducerInterceptor.class);
+ ProducerImpl<byte[]> producer = constructProducer(client, new ProducerConfigurationData(),
+ new ProducerInterceptors(Collections.singletonList(interceptor)));
+
+ producer.closeAsync().get(5, TimeUnit.SECONDS);
+ producer.closeAsync().get(5, TimeUnit.SECONDS);
+
+ verify(interceptor).close();
+ }
+
private ProducerConfigurationData encryptedProducerConf() {
ProducerConfigurationData conf = new ProducerConfigurationData();
conf.setEncryptionKeys(new TreeSet<>(Collections.singleton("key")));
@@ -336,12 +351,17 @@ public class ProducerImplTest {
* {@code msgCrypto} fields hold real values instead of needing reflection: {@code conf.setMessageCrypto()}
* installs a test crypto through the same constructor branch production uses.
*/
- @SuppressWarnings("unchecked")
private static ProducerImpl<byte[]> constructProducer(PulsarClientImpl client,
ProducerConfigurationData conf) {
+ return constructProducer(client, conf, null);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static ProducerImpl<byte[]> constructProducer(PulsarClientImpl client,
+ ProducerConfigurationData conf, ProducerInterceptors interceptors) {
return mock(ProducerImpl.class, withSettings()
.useConstructor(client, "persistent://public/default/producer-impl-test", conf,
- new CompletableFuture<>(), 0, Schema.BYTES, null, Optional.empty())
+ new CompletableFuture<>(), 0, Schema.BYTES, interceptors, Optional.empty())
.defaultAnswer(CALLS_REAL_METHODS));
}