sijie closed pull request #2471: PIP-23: Pulsar Java Client Interceptors.
URL: https://github.com/apache/incubator-pulsar/pull/2471
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
index ae1a4dba17..99aec9355b 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
@@ -111,7 +111,7 @@ public String toString() {
consumerFuture,
SubscriptionMode.Durable,
MessageId.earliest,
- Schema.BYTES);
+ Schema.BYTES, null);
incomingRawMessages = new GrowableArrayBlockingQueue<>();
pendingRawReceives = new ConcurrentLinkedQueue<>();
}
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
index f8ac40a945..8c175d7326 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
@@ -24,6 +24,7 @@
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyObject;
import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.matches;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.doAnswer;
@@ -36,7 +37,6 @@
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
-import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
@@ -80,7 +80,6 @@
import org.apache.pulsar.broker.cache.ConfigurationCacheService;
import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService;
import org.apache.pulsar.broker.namespace.NamespaceService;
-import org.apache.pulsar.broker.service.Topic.PublishContext;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentReplicator;
import org.apache.pulsar.broker.service.persistent.CompactorSubscription;
import
org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
@@ -1224,7 +1223,7 @@ public void testClosingReplicationProducerTwice() throws
Exception {
verify(clientImpl)
.createProducerAsync(
any(ProducerConfigurationData.class),
- any(Schema.class)
+ any(Schema.class), eq(null)
);
replicator.disconnect(false);
@@ -1235,7 +1234,7 @@ public void testClosingReplicationProducerTwice() throws
Exception {
verify(clientImpl, Mockito.times(2))
.createProducerAsync(
any(ProducerConfigurationData.class),
- any(Schema.class)
+ any(Schema.class), any(null)
);
}
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
index de4992594d..528ff64d7e 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.service;
+import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
@@ -242,7 +243,7 @@ public void testConcurrentReplicator() throws Exception {
Thread.sleep(3000);
Mockito.verify(pulsarClient,
Mockito.times(1)).createProducerAsync(Mockito.any(ProducerConfigurationData.class),
- Mockito.any(Schema.class));
+ Mockito.any(Schema.class), eq(null));
client1.shutdown();
}
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/v1/V1_ReplicatorTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/v1/V1_ReplicatorTest.java
index 0d6d3036ef..92bb64175a 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/v1/V1_ReplicatorTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/v1/V1_ReplicatorTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.service.v1;
+import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
@@ -241,7 +242,7 @@ public void testConcurrentReplicator() throws Exception {
Thread.sleep(3000);
Mockito.verify(pulsarClient,
Mockito.times(1)).createProducerAsync(Mockito.any(ProducerConfigurationData.class),
- Mockito.any(Schema.class));
+ Mockito.any(Schema.class), eq(null));
client1.shutdown();
}
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
new file mode 100644
index 0000000000..d83384698e
--- /dev/null
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java
@@ -0,0 +1,359 @@
+/**
+ * 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.api;
+
+import org.apache.pulsar.client.impl.MessageImpl;
+import org.apache.pulsar.client.impl.TopicMessageImpl;
+import org.apache.pulsar.common.api.proto.PulsarApi;
+import org.junit.Assert;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class InterceptorsTest extends ProducerConsumerBase {
+
+ private static final Logger log =
LoggerFactory.getLogger(InterceptorsTest.class);
+
+ @BeforeMethod
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @AfterMethod
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test
+ public void testProducerInterceptor() throws PulsarClientException {
+ ProducerInterceptor<String> interceptor1 = new
ProducerInterceptor<String>() {
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public Message<String> beforeSend(Producer<String> producer,
Message<String> message) {
+ MessageImpl<String> msg = (MessageImpl<String>) message;
+ log.info("Before send message: {}", new String(msg.getData()));
+
java.util.List<org.apache.pulsar.common.api.proto.PulsarApi.KeyValue>
properties = msg.getMessageBuilder().getPropertiesList();
+ for (int i = 0; i < properties.size(); i++) {
+ if ("key".equals(properties.get(i).getKey())) {
+ msg.getMessageBuilder().setProperties(i,
PulsarApi.KeyValue.newBuilder().setKey("key").setValue("after").build());
+ }
+ }
+ return msg;
+ }
+
+ @Override
+ public void onSendAcknowledgement(Producer<String> producer,
Message<String> message, MessageId msgId, Throwable cause) {
+ message.getProperties();
+ Assert.assertEquals("complete", message.getProperty("key"));
+ log.info("Send acknowledgement message: {}, msgId: {}", new
String(message.getData()), msgId, cause);
+ }
+ };
+
+ ProducerInterceptor<String> interceptor2 = new
ProducerInterceptor<String>() {
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public Message<String> beforeSend(Producer<String> producer,
Message<String> message) {
+ MessageImpl<String> msg = (MessageImpl<String>) message;
+ log.info("Before send message: {}", new String(msg.getData()));
+
java.util.List<org.apache.pulsar.common.api.proto.PulsarApi.KeyValue>
properties = msg.getMessageBuilder().getPropertiesList();
+ for (int i = 0; i < properties.size(); i++) {
+ if ("key".equals(properties.get(i).getKey())) {
+ msg.getMessageBuilder().setProperties(i,
PulsarApi.KeyValue.newBuilder().setKey("key").setValue("complete").build());
+ }
+ }
+ return msg;
+ }
+
+ @Override
+ public void onSendAcknowledgement(Producer<String> producer,
Message<String> message, MessageId msgId, Throwable cause) {
+ message.getProperties();
+ Assert.assertEquals("complete", message.getProperty("key"));
+ log.info("Send acknowledgement message: {}, msgId: {}", new
String(message.getData()), msgId, cause);
+ }
+ };
+
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic("persistent://my-property/my-ns/my-topic")
+ .intercept(interceptor1, interceptor2)
+ .create();
+
+ MessageId messageId = producer.newMessage().property("key",
"before").value("Hello Pulsar!").send();
+ log.info("Send result messageId: {}", messageId);
+ producer.close();
+ }
+
+ @Test
+ public void testConsumerInterceptorWithSingleTopicSubscribe() throws
PulsarClientException {
+ ConsumerInterceptor<String> interceptor = new
ConsumerInterceptor<String>() {
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public Message<String> beforeConsume(Consumer<String> consumer,
Message<String> message) {
+ MessageImpl<String> msg = (MessageImpl<String>) message;
+
msg.getMessageBuilder().addProperties(PulsarApi.KeyValue.newBuilder().setKey("beforeConsumer").setValue("1").build());
+ return msg;
+ }
+
+ @Override
+ public void onAcknowledge(Consumer<String> consumer, MessageId
messageId, Throwable cause) {
+ log.info("onAcknowledge messageId: {}", messageId, cause);
+ }
+
+ @Override
+ public void onAcknowledgeCumulative(Consumer<String> consumer,
MessageId messageId, Throwable cause) {
+ log.info("onAcknowledgeCumulative messageIds: {}", messageId,
cause);
+ }
+ };
+
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic("persistent://my-property/my-ns/my-topic")
+ .subscriptionType(SubscriptionType.Shared)
+ .intercept(interceptor)
+ .subscriptionName("my-subscription")
+ .subscribe();
+
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic("persistent://my-property/my-ns/my-topic")
+ .create();
+
+ producer.newMessage().value("Hello Pulsar!").send();
+
+ Message<String> received = consumer.receive();
+ MessageImpl<String> msg = (MessageImpl<String>) received;
+ boolean haveKey = false;
+ for (PulsarApi.KeyValue keyValue :
msg.getMessageBuilder().getPropertiesList()) {
+ if ("beforeConsumer".equals(keyValue.getKey())) {
+ haveKey = true;
+ }
+ }
+ Assert.assertTrue(haveKey);
+ consumer.acknowledge(received);
+ producer.close();
+ consumer.close();
+ }
+
+ @Test
+ public void testConsumerInterceptorWithMultiTopicSubscribe() throws
PulsarClientException {
+
+ ConsumerInterceptor<String> interceptor = new
ConsumerInterceptor<String>() {
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public Message<String> beforeConsume(Consumer<String> consumer,
Message<String> message) {
+ MessageImpl<String> msg = (MessageImpl<String>) message;
+
msg.getMessageBuilder().addProperties(PulsarApi.KeyValue.newBuilder().setKey("beforeConsumer").setValue("1").build());
+ return msg;
+ }
+
+ @Override
+ public void onAcknowledge(Consumer<String> consumer, MessageId
messageId, Throwable cause) {
+ log.info("onAcknowledge messageId: {}", messageId, cause);
+ }
+
+ @Override
+ public void onAcknowledgeCumulative(Consumer<String> consumer,
MessageId messageId, Throwable cause) {
+ log.info("onAcknowledgeCumulative messageIds: {}", messageId,
cause);
+ }
+ };
+
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic("persistent://my-property/my-ns/my-topic")
+ .create();
+
+ Producer<String> producer1 = pulsarClient.newProducer(Schema.STRING)
+ .topic("persistent://my-property/my-ns/my-topic1")
+ .create();
+
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic("persistent://my-property/my-ns/my-topic",
"persistent://my-property/my-ns/my-topic1")
+ .subscriptionType(SubscriptionType.Shared)
+ .intercept(interceptor)
+ .subscriptionName("my-subscription")
+ .subscribe();
+
+ producer.newMessage().value("Hello Pulsar!").send();
+ producer1.newMessage().value("Hello Pulsar!").send();
+
+ int keyCount = 0;
+ for (int i = 0; i < 2; i++) {
+ Message<String> received = consumer.receive();
+ MessageImpl<String> msg = (MessageImpl<String>)
((TopicMessageImpl<String>) received).getMessage();
+ for (PulsarApi.KeyValue keyValue :
msg.getMessageBuilder().getPropertiesList()) {
+ if ("beforeConsumer".equals(keyValue.getKey())) {
+ keyCount++;
+ }
+ }
+ consumer.acknowledge(received);
+ }
+ Assert.assertEquals(2, keyCount);
+ producer.close();
+ producer1.close();
+ consumer.close();
+ }
+
+ @Test
+ public void testConsumerInterceptorWithPatternTopicSubscribe() throws
PulsarClientException {
+
+ ConsumerInterceptor<String> interceptor = new
ConsumerInterceptor<String>() {
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public Message<String> beforeConsume(Consumer<String> consumer,
Message<String> message) {
+ MessageImpl<String> msg = (MessageImpl<String>) message;
+
msg.getMessageBuilder().addProperties(PulsarApi.KeyValue.newBuilder().setKey("beforeConsumer").setValue("1").build());
+ return msg;
+ }
+
+ @Override
+ public void onAcknowledge(Consumer<String> consumer, MessageId
messageId, Throwable cause) {
+ log.info("onAcknowledge messageId: {}", messageId, cause);
+ }
+
+ @Override
+ public void onAcknowledgeCumulative(Consumer<String> consumer,
MessageId messageId, Throwable cause) {
+ log.info("onAcknowledgeCumulative messageIds: {}", messageId,
cause);
+ }
+ };
+
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic("persistent://my-property/my-ns/my-topic")
+ .create();
+
+ Producer<String> producer1 = pulsarClient.newProducer(Schema.STRING)
+ .topic("persistent://my-property/my-ns/my-topic1")
+ .create();
+
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topicsPattern("persistent://my-property/my-ns/my-.*")
+ .subscriptionType(SubscriptionType.Shared)
+ .intercept(interceptor)
+ .subscriptionName("my-subscription")
+ .subscribe();
+
+ producer.newMessage().value("Hello Pulsar!").send();
+ producer1.newMessage().value("Hello Pulsar!").send();
+
+ int keyCount = 0;
+ for (int i = 0; i < 2; i++) {
+ Message<String> received = consumer.receive();
+ MessageImpl<String> msg = (MessageImpl<String>)
((TopicMessageImpl<String>) received).getMessage();
+ for (PulsarApi.KeyValue keyValue :
msg.getMessageBuilder().getPropertiesList()) {
+ if ("beforeConsumer".equals(keyValue.getKey())) {
+ keyCount++;
+ }
+ }
+ consumer.acknowledge(received);
+ }
+ Assert.assertEquals(2, keyCount);
+ producer.close();
+ producer1.close();
+ consumer.close();
+ }
+
+ @Test
+ public void testConsumerInterceptorForAcknowledgeCumulative() throws
PulsarClientException {
+
+ List<MessageId> ackHolder = new ArrayList<>();
+
+ ConsumerInterceptor<String> interceptor = new
ConsumerInterceptor<String>() {
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public Message<String> beforeConsume(Consumer<String> consumer,
Message<String> message) {
+ MessageImpl<String> msg = (MessageImpl<String>) message;
+
msg.getMessageBuilder().addProperties(PulsarApi.KeyValue.newBuilder().setKey("beforeConsumer").setValue("1").build());
+ return msg;
+ }
+
+ @Override
+ public void onAcknowledge(Consumer<String> consumer, MessageId
messageId, Throwable cause) {
+ log.info("onAcknowledge messageId: {}", messageId, cause);
+ }
+
+ @Override
+ public void onAcknowledgeCumulative(Consumer<String> consumer,
MessageId messageId, Throwable cause) {
+ long acknowledged = ackHolder.stream().filter(m ->
(m.compareTo(messageId) <= 0)).count();
+ Assert.assertEquals(acknowledged, 100);
+ ackHolder.clear();
+ log.info("onAcknowledgeCumulative messageIds: {}", messageId,
cause);
+ }
+ };
+
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic("persistent://my-property/my-ns/my-topic")
+ .subscriptionType(SubscriptionType.Failover)
+ .intercept(interceptor)
+ .subscriptionName("my-subscription")
+ .subscribe();
+
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic("persistent://my-property/my-ns/my-topic")
+ .create();
+
+ for (int i = 0; i < 100; i++) {
+ producer.newMessage().value("Hello Pulsar!").send();
+ }
+
+ int keyCount = 0;
+ for (int i = 0; i < 100; i++) {
+ Message<String> received = consumer.receive();
+ MessageImpl<String> msg = (MessageImpl<String>) received;
+ for (PulsarApi.KeyValue keyValue :
msg.getMessageBuilder().getPropertiesList()) {
+ if ("beforeConsumer".equals(keyValue.getKey())) {
+ keyCount++;
+ }
+ }
+ ackHolder.add(received.getMessageId());
+ if (i == 99) {
+ consumer.acknowledgeCumulative(received);
+ }
+ }
+ Assert.assertEquals(100, keyCount);
+ producer.close();
+ consumer.close();
+ }
+}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/Consumer.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/Consumer.java
index 69d885fb78..f5a926602d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/Consumer.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/Consumer.java
@@ -288,4 +288,10 @@
* @return Whether the consumer is connected to the broker
*/
boolean isConnected();
+
+ /**
+ * Get the name of consumer.
+ * @return consumer name.
+ */
+ String getConsumerName();
}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
index 8657859103..da1cbb5d17 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
@@ -330,4 +330,12 @@
* Set subscriptionInitialPosition for the consumer
*/
ConsumerBuilder<T> subscriptionInitialPosition(SubscriptionInitialPosition
subscriptionInitialPosition);
+
+ /**
+ * Intercept {@link Consumer}.
+ *
+ * @param interceptors the list of interceptors to intercept the consumer
created by this builder.
+ * @return consumer builder.
+ */
+ ConsumerBuilder<T> intercept(ConsumerInterceptor<T> ...interceptors);
}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerInterceptor.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerInterceptor.java
new file mode 100644
index 0000000000..1134d8a2b4
--- /dev/null
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerInterceptor.java
@@ -0,0 +1,99 @@
+/**
+ * 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.api;
+
+import java.util.List;
+
+/**
+ * A plugin interface that allows you to intercept (and possibly mutate)
+ * messages received by the consumer.
+ * <p>
+ * A primary use case is to hook into consumer applications for custom
+ * monitoring, logging, etc.
+ * <p>
+ * Exceptions thrown by interceptor methods will be caught, logged, but
+ * not propagated further.
+ */
+public interface ConsumerInterceptor<T> extends AutoCloseable {
+
+ /**
+ * Close the interceptor.
+ */
+ void close();
+
+ /**
+ * This is called just before the message is returned by
+ * {@link Consumer#receive()}, {@link MessageListener#received(Consumer,
+ * Message)} or the {@link java.util.concurrent.CompletableFuture}
returned by
+ * {@link Consumer#receiveAsync()} completes.
+ * <p>
+ * This method is allowed to modify message, in which case the new message
+ * will be returned.
+ * <p>
+ * Any exception thrown by this method will be caught by the caller,
logged,
+ * but not propagated to client.
+ * <p>
+ * Since the consumer may run multiple interceptors, a particular
+ * interceptor's
+ * <tt>beforeConsume</tt> callback will be called in the order specified by
+ * {@link ConsumerBuilder#intercept(ConsumerInterceptor[])}. The first
+ * interceptor in the list gets the consumed message, the following
+ * interceptor will be passed
+ * the message returned by the previous interceptor, and so on. Since
+ * interceptors are allowed to modify message, interceptors may potentially
+ * get the messages already modified by other interceptors. However
building a
+ * pipeline of mutable
+ * interceptors that depend on the output of the previous interceptor is
+ * discouraged, because of potential side-effects caused by interceptors
+ * potentially failing to modify the message and throwing an exception.
+ * if one of interceptors in the list throws an exception from
+ * <tt>beforeConsume</tt>, the exception is caught, logged,
+ * and the next interceptor is called with the message returned by the last
+ * successful interceptor in the list, or otherwise the original consumed
+ * message.
+ *
+ * @param consumer the consumer which contains the interceptor
+ * @param message the message to be consumed by the client.
+ * @return message that is either modified by the interceptor or same
message
+ * passed into the method.
+ */
+ Message<T> beforeConsume(Consumer<T> consumer, Message<T> message);
+
+ /**
+ * This is called consumer sends the acknowledgment to the broker.
+ *
+ * <p>Any exception thrown by this method will be ignored by the caller.
+ *
+ * @param consumer the consumer which contains the interceptor
+ * @param messageId message to ack, null if acknowledge fail.
+ * @param exception the exception on acknowledge.
+ */
+ void onAcknowledge(Consumer<T> consumer, MessageId messageId, Throwable
exception);
+
+ /**
+ * This is called consumer send the cumulative acknowledgment to the
broker.
+ *
+ * <p>Any exception thrown by this method will be ignored by the caller.
+ *
+ * @param consumer the consumer which contains the interceptor
+ * @param messageId message to ack, null if acknowledge fail.
+ * @param exception the exception on acknowledge.
+ */
+ void onAcknowledgeCumulative(Consumer<T> consumer, MessageId messageId,
Throwable exception);
+}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
index 8256b4a282..b3aa720902 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
@@ -318,4 +318,12 @@
* @return
*/
ProducerBuilder<T> properties(Map<String, String> properties);
+
+ /**
+ * Intercept {@link Producer}.
+ *
+ * @param interceptors the list of interceptors to intercept the producer
created by this builder.
+ * @return producer builder.
+ */
+ ProducerBuilder<T> intercept(ProducerInterceptor<T> ... interceptors);
}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerInterceptor.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerInterceptor.java
new file mode 100644
index 0000000000..b6a0d77558
--- /dev/null
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerInterceptor.java
@@ -0,0 +1,92 @@
+/**
+ * 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.api;
+
+/**
+ * A plugin interface that allows you to intercept (and possibly mutate) the
+ * messages received by the producer before they are published to the Pulsar
+ * brokers.
+ * <p>
+ * Exceptions thrown by ProducerInterceptor methods will be caught, logged, but
+ * not propagated further.
+ * <p>
+ * ProducerInterceptor callbacks may be called from multiple threads.
Interceptor
+ * implementation must ensure thread-safety, if needed.
+ */
+public interface ProducerInterceptor<T> extends AutoCloseable {
+
+ /**
+ * Close the interceptor.
+ */
+ void close();
+
+ /**
+ * This is called from {@link Producer#send(Object)} and {@link
+ * Producer#sendAsync(Object)} methods, before
+ * send the message to the brokers. This method is allowed to modify the
+ * record, in which case, the new record
+ * will be returned.
+ * <p>
+ * Any exception thrown by this method will be caught by the caller and
+ * logged, but not propagated further.
+ * <p>
+ * Since the producer may run multiple interceptors, a particular
+ * interceptor's {@link #beforeSend(Producer, Message)} callback will be
called in the
+ * order specified by
+ * {@link ProducerBuilder#intercept(ProducerInterceptor[])}.
+ * <p>
+ * The first interceptor in the list gets the message passed from the
client,
+ * the following interceptor will be passed the message returned by the
+ * previous interceptor, and so on. Since interceptors are allowed to
modify
+ * messages, interceptors may potentially get the message already modified
by
+ * other interceptors. However, building a pipeline of mutable interceptors
+ * that depend on the output of the previous interceptor is discouraged,
+ * because of potential side-effects caused by interceptors potentially
+ * failing to modify the message and throwing an exception. If one of the
+ * interceptors in the list throws an exception from
+ * {@link#beforeSend(Message)}, the exception is caught, logged, and the
next
+ * interceptor is called with the message returned by the last successful
+ * interceptor in the list, or otherwise the client.
+ *
+ * @param producer the producer which contains the interceptor.
+ * @param message message to send
+ * @return the intercepted message
+ */
+ Message<T> beforeSend(Producer<T> producer, Message<T> message);
+
+ /**
+ * This method is called when the message sent to the broker has been
+ * acknowledged, or when sending the message fails.
+ * This method is generally called just before the user callback is
+ * called, and in additional cases when an exception on the producer side.
+ * <p>
+ * Any exception thrown by this method will be ignored by the caller.
+ * <p>
+ * This method will generally execute in the background I/O thread, so the
+ * implementation should be reasonably fast. Otherwise, sending of messages
+ * from other threads could be delayed.
+ *
+ * @param producer the producer which contains the interceptor.
+ * @param message the message that application sends
+ * @param msgId the message id that assigned by the broker; null if send
failed.
+ * @param exception the exception on sending messages, null indicates send
has succeed.
+ */
+ void onSendAcknowledgement(Producer<T> producer, Message<T> message,
MessageId msgId, Throwable exception);
+
+}
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 a6fda6f8b6..d8a0de3b9a 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
@@ -60,10 +60,11 @@
protected final ConcurrentLinkedQueue<CompletableFuture<Message<T>>>
pendingReceives;
protected int maxReceiverQueueSize;
protected Schema<T> schema;
+ protected final ConsumerInterceptors<T> interceptors;
protected ConsumerBase(PulsarClientImpl client, String topic,
ConsumerConfigurationData<T> conf,
int receiverQueueSize, ExecutorService
listenerExecutor,
- CompletableFuture<Consumer<T>> subscribeFuture,
Schema<T> schema) {
+ CompletableFuture<Consumer<T>> subscribeFuture,
Schema<T> schema, ConsumerInterceptors interceptors) {
super(client, topic);
this.maxReceiverQueueSize = receiverQueueSize;
this.subscription = conf.getSubscriptionName();
@@ -81,6 +82,7 @@ protected ConsumerBase(PulsarClientImpl client, String topic,
ConsumerConfigurat
this.listenerExecutor = listenerExecutor;
this.pendingReceives = Queues.newConcurrentLinkedQueue();
this.schema = schema;
+ this.interceptors = interceptors;
}
@Override
@@ -335,6 +337,7 @@ public String getSubscription() {
return subscription;
}
+ @Override
public String getConsumerName() {
return this.consumerName;
}
@@ -360,4 +363,24 @@ protected void setMaxReceiverQueueSize(int newSize) {
this.maxReceiverQueueSize = newSize;
}
+ protected Message<T> beforeConsume(Message<T> message) {
+ if (interceptors != null) {
+ return interceptors.beforeConsume(this, message);
+ } else {
+ return message;
+ }
+ }
+
+ protected void onAcknowledge(MessageId messageId, Throwable exception) {
+ if (interceptors != null) {
+ interceptors.onAcknowledge(this, messageId, exception);
+ }
+ }
+
+ protected void onAcknowledgeCumulative(MessageId messageId, Throwable
exception) {
+ if (interceptors != null) {
+ interceptors.onAcknowledgeCumulative(this, messageId, exception);
+ }
+ }
+
}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
index f0067f7f10..2095babbb1 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
@@ -20,6 +20,8 @@
import static com.google.common.base.Preconditions.checkArgument;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -32,6 +34,7 @@
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
import org.apache.pulsar.client.api.ConsumerEventListener;
+import org.apache.pulsar.client.api.ConsumerInterceptor;
import org.apache.pulsar.client.api.CryptoKeyReader;
import org.apache.pulsar.client.api.MessageListener;
import org.apache.pulsar.client.api.PulsarClientException;
@@ -52,6 +55,7 @@
private final PulsarClientImpl client;
private ConsumerConfigurationData<T> conf;
private final Schema<T> schema;
+ private List<ConsumerInterceptor<T>> interceptorList;
private static long MIN_ACK_TIMEOUT_MILLIS = 1000;
@@ -104,8 +108,9 @@
return FutureUtil.failedFuture(
new InvalidConfigurationException("Subscription name must
be set on the consumer builder"));
}
-
- return client.subscribeAsync(conf, schema);
+ return interceptorList == null || interceptorList.size() == 0 ?
+ client.subscribeAsync(conf, schema, null) :
+ client.subscribeAsync(conf, schema, new
ConsumerInterceptors<>(interceptorList));
}
@Override
@@ -242,7 +247,16 @@
return this;
}
- public ConsumerConfigurationData<T> getConf() {
+ @Override
+ public ConsumerBuilder<T> intercept(ConsumerInterceptor<T>...
interceptors) {
+ if (interceptorList == null) {
+ interceptorList = new ArrayList<>();
+ }
+ interceptorList.addAll(Arrays.asList(interceptors));
+ return this;
+ }
+
+ public ConsumerConfigurationData<T> getConf() {
return conf;
}
}
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 fe37b69a50..a0a231902f 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
@@ -143,14 +143,14 @@
}
ConsumerImpl(PulsarClientImpl client, String topic,
ConsumerConfigurationData<T> conf,
- ExecutorService listenerExecutor, int partitionIndex,
CompletableFuture<Consumer<T>> subscribeFuture, Schema<T> schema) {
- this(client, topic, conf, listenerExecutor, partitionIndex,
subscribeFuture, SubscriptionMode.Durable, null, schema);
+ ExecutorService listenerExecutor, int partitionIndex,
CompletableFuture<Consumer<T>> subscribeFuture, Schema<T> schema,
ConsumerInterceptors interceptors) {
+ this(client, topic, conf, listenerExecutor, partitionIndex,
subscribeFuture, SubscriptionMode.Durable, null, schema, interceptors);
}
ConsumerImpl(PulsarClientImpl client, String topic,
ConsumerConfigurationData<T> conf,
ExecutorService listenerExecutor, int partitionIndex,
CompletableFuture<Consumer<T>> subscribeFuture,
- SubscriptionMode subscriptionMode, MessageId startMessageId,
Schema<T> schema) {
- super(client, topic, conf, conf.getReceiverQueueSize(),
listenerExecutor, subscribeFuture, schema);
+ SubscriptionMode subscriptionMode, MessageId startMessageId,
Schema<T> schema, ConsumerInterceptors interceptors) {
+ super(client, topic, conf, conf.getReceiverQueueSize(),
listenerExecutor, subscribeFuture, schema, interceptors);
this.consumerId = client.newConsumerId();
this.subscriptionMode = subscriptionMode;
this.startMessageId = startMessageId != null ? new
BatchMessageIdImpl((MessageIdImpl) startMessageId) : null;
@@ -263,8 +263,9 @@ public UnAckedMessageTracker getUnAckedMessageTracker() {
Message<T> message;
try {
message = incomingMessages.take();
- messageProcessed(message);
- return message;
+ Message<T> interceptMsg = beforeConsume(message);
+ messageProcessed(interceptMsg);
+ return interceptMsg;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
stats.incrementNumReceiveFailed();
@@ -293,8 +294,9 @@ public UnAckedMessageTracker getUnAckedMessageTracker() {
if (message == null && conf.getReceiverQueueSize() == 0) {
sendFlowPermitsToBroker(cnx(), 1);
} else if (message != null) {
- messageProcessed(message);
- result.complete(message);
+ Message<T> interceptMsg = beforeConsume(message);
+ messageProcessed(interceptMsg);
+ result.complete(interceptMsg);
}
return result;
@@ -352,10 +354,11 @@ public UnAckedMessageTracker getUnAckedMessageTracker() {
Message<T> message;
try {
message = incomingMessages.poll(timeout, unit);
- if (message != null) {
- messageProcessed(message);
+ Message<T> interceptMsg = beforeConsume(message);
+ if (interceptMsg != null) {
+ messageProcessed(interceptMsg);
}
- return message;
+ return interceptMsg;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -410,7 +413,13 @@ boolean markAckForBatchMessage(BatchMessageIdImpl
batchMessageId, AckType ackTyp
checkArgument(messageId instanceof MessageIdImpl);
if (getState() != State.Ready && getState() != State.Connecting) {
stats.incrementNumAcksFailed();
- return FutureUtil.failedFuture(new PulsarClientException("Consumer
not ready. State: " + getState()));
+ PulsarClientException exception = new
PulsarClientException("Consumer not ready. State: " + getState());
+ if (AckType.Individual.equals(ackType)) {
+ onAcknowledge(messageId, exception);
+ } else if (AckType.Cumulative.equals(ackType)) {
+ onAcknowledgeCumulative(messageId, exception);
+ }
+ return FutureUtil.failedFuture(exception);
}
if (messageId instanceof BatchMessageIdImpl) {
@@ -444,7 +453,9 @@ boolean markAckForBatchMessage(BatchMessageIdImpl
batchMessageId, AckType ackTyp
unAckedMessageTracker.remove(msgId);
stats.incrementNumAcksSent(1);
}
+ onAcknowledge(messageId, null);
} else if (ackType == AckType.Cumulative) {
+ onAcknowledgeCumulative(messageId, null);
stats.incrementNumAcksSent(unAckedMessageTracker.removeMessagesTill(msgId));
}
@@ -837,9 +848,10 @@ void notifyPendingReceivedCallback(final Message<T>
message, Exception exception
receivedFuture.complete(message);
} else {
// increase permits for available message-queue
- messageProcessed(message);
+ Message<T> interceptMsg = beforeConsume(message);
+ messageProcessed(interceptMsg);
// return message to receivedCallback
- listenerExecutor.execute(() ->
receivedFuture.complete(message));
+ listenerExecutor.execute(() ->
receivedFuture.complete(interceptMsg));
}
}
} 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
new file mode 100644
index 0000000000..a0d30fcb96
--- /dev/null
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerInterceptors.java
@@ -0,0 +1,132 @@
+/**
+ * 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 org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerInterceptor;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * A container that hold the list {@link
org.apache.pulsar.client.api.ConsumerInterceptor} and wraps calls to the chain
+ * of custom interceptors.
+ */
+public class ConsumerInterceptors<T> implements Closeable {
+
+ private static final Logger log =
LoggerFactory.getLogger(ConsumerInterceptors.class);
+
+ private final List<ConsumerInterceptor<T>> interceptors;
+
+ public ConsumerInterceptors(List<ConsumerInterceptor<T>> interceptors) {
+ this.interceptors = interceptors;
+ }
+
+ /**
+ * This is called just before the message is returned by {@link
Consumer#receive()},
+ * {@link MessageListener#received(Consumer, Message)} or the {@link
java.util.concurrent.CompletableFuture}
+ * returned by {@link Consumer#receiveAsync()} completes.
+ * <p>
+ * This method calls {@link ConsumerInterceptor#beforeConsume(Consumer,
Message)} for each interceptor. Messages returned
+ * from each interceptor get passed to beforeConsume() of the next
interceptor in the chain of interceptors.
+ * <p>
+ * This method does not throw exceptions. If any of the interceptors in
the chain throws an exception, it gets
+ * caught and logged, and next interceptor in int the chain is called with
'messages' returned by the previous
+ * successful interceptor beforeConsume call.
+ *
+ * @param consumer the consumer which contains the interceptors
+ * @param message message to be consume by the client.
+ * @return messages that are either modified by interceptors or same as
messages passed to this method.
+ */
+ public Message<T> beforeConsume(Consumer<T> consumer, Message<T> message) {
+ Message<T> interceptorMessage = message;
+ for (int i = 0; i < interceptors.size(); i++) {
+ try {
+ interceptorMessage =
interceptors.get(i).beforeConsume(consumer, interceptorMessage);
+ } catch (Exception e) {
+ if (consumer != null) {
+ log.warn("Error executing interceptor beforeConsume
callback topic: {} consumerName: {}", consumer.getTopic(),
consumer.getConsumerName(), e);
+ } else {
+ log.warn("Error executing interceptor beforeConsume
callback", e);
+ }
+ }
+ }
+ return interceptorMessage;
+ }
+
+ /**
+ * This is called when acknowledge request return from the broker.
+ * <p>
+ * This method calls {@link ConsumerInterceptor#onAcknowledge(Consumer,
MessageId, Throwable)} method for each interceptor.
+ * <p>
+ * This method does not throw exceptions. Exceptions thrown by any of
interceptors in the chain are logged, but not propagated.
+ *
+ * @param consumer the consumer which contains the interceptors
+ * @param messageId message to acknowledge.
+ * @param exception exception returned by broker.
+ */
+ public void onAcknowledge(Consumer<T> consumer, MessageId messageId,
Throwable exception) {
+ for (int i = 0; i < interceptors.size(); i++) {
+ try {
+ interceptors.get(i).onAcknowledge(consumer, messageId,
exception);
+ } catch (Exception e) {
+ log.warn("Error executing interceptor onAcknowledge callback
", e);
+ }
+ }
+ }
+
+ /**
+ * This is called when acknowledge cumulative request return from the
broker.
+ * <p>
+ * This method calls {@link
ConsumerInterceptor#onAcknowledgeCumulative(Consumer, MessageId, Throwable)}
(Message, Throwable)} method for each interceptor.
+ * <p>
+ * This method does not throw exceptions. Exceptions thrown by any of
interceptors in the chain are logged, but not propagated.
+ *
+ * @param consumer the consumer which contains the interceptors
+ * @param messageId messages to acknowledge.
+ * @param exception exception returned by broker.
+ */
+ public void onAcknowledgeCumulative(Consumer<T> consumer, MessageId
messageId, Throwable exception) {
+ for (int i = 0; i < interceptors.size(); i++) {
+ try {
+ interceptors.get(i).onAcknowledgeCumulative(consumer,
messageId, exception);
+ } catch (Exception e) {
+ log.warn("Error executing interceptor onAcknowledgeCumulative
callback ", e);
+ }
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ for (int i = 0; i < interceptors.size(); i++) {
+ try {
+ interceptors.get(i).close();
+ } catch (Exception e) {
+ log.error("Fail to close consumer interceptor ", e);
+ }
+ }
+ }
+
+}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
index 96b68c1af4..97e2247f8d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
@@ -241,7 +241,7 @@ public String getProducerName() {
return null;
}
- ByteBuf getDataBuffer() {
+ public ByteBuf getDataBuffer() {
return payload;
}
@@ -274,7 +274,7 @@ public String getProperty(String name) {
return properties.get(name);
}
- MessageMetadata.Builder getMessageBuilder() {
+ public MessageMetadata.Builder getMessageBuilder() {
return msgMetadataBuilder;
}
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 f1cb9cf262..75fdac681a 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
@@ -85,9 +85,9 @@
private final ConsumerConfigurationData<T> internalConfig;
MultiTopicsConsumerImpl(PulsarClientImpl client,
ConsumerConfigurationData<T> conf, ExecutorService listenerExecutor,
- CompletableFuture<Consumer<T>> subscribeFuture,
Schema<T> schema) {
+ CompletableFuture<Consumer<T>> subscribeFuture,
Schema<T> schema, ConsumerInterceptors<T> interceptors) {
super(client, "TopicsConsumerFakeTopicName" +
ConsumerName.generateRandomName(), conf,
- Math.max(2, conf.getReceiverQueueSize()), listenerExecutor,
subscribeFuture, schema);
+ Math.max(2, conf.getReceiverQueueSize()), listenerExecutor,
subscribeFuture, schema, interceptors);
checkArgument(conf.getReceiverQueueSize() > 0,
"Receiver queue size needs to be greater than 0 for Topics
Consumer");
@@ -632,7 +632,7 @@ private boolean topicNameValid(String topicName) {
ExecutorService listenerExecutor,
CompletableFuture<Consumer<T>> subscribeFuture,
int
numPartitions,
-
Schema<T> schema) {
+
Schema<T> schema, ConsumerInterceptors<T> interceptors) {
checkArgument(conf.getTopicNames().size() == 1, "Should have only 1
topic for partitioned consumer");
// get topic name, then remove it from conf, so constructor will
create a consumer with no topic.
@@ -641,7 +641,7 @@ private boolean topicNameValid(String topicName) {
cloneConf.getTopicNames().remove(topicName);
CompletableFuture<Consumer> future = new CompletableFuture<>();
- MultiTopicsConsumerImpl consumer = new MultiTopicsConsumerImpl(client,
cloneConf, listenerExecutor, future, schema);
+ MultiTopicsConsumerImpl consumer = new MultiTopicsConsumerImpl(client,
cloneConf, listenerExecutor, future, schema, interceptors);
future.thenCompose(c ->
((MultiTopicsConsumerImpl)c).subscribeAsync(topicName, numPartitions))
.thenRun(()-> subscribeFuture.complete(consumer))
@@ -695,7 +695,7 @@ private void
subscribeTopicPartitions(CompletableFuture<Void> subscribeResult, S
String partitionName =
TopicName.get(topicName).getPartition(partitionIndex).toString();
CompletableFuture<Consumer<T>> subFuture = new
CompletableFuture<>();
ConsumerImpl<T> newConsumer = new
ConsumerImpl<>(client, partitionName, configurationData,
- client.externalExecutorProvider().getExecutor(),
partitionIndex, subFuture, schema);
+ client.externalExecutorProvider().getExecutor(),
partitionIndex, subFuture, schema, interceptors);
consumers.putIfAbsent(newConsumer.getTopic(),
newConsumer);
return subFuture;
})
@@ -706,7 +706,7 @@ private void
subscribeTopicPartitions(CompletableFuture<Void> subscribeResult, S
CompletableFuture<Consumer<T>> subFuture = new
CompletableFuture<>();
ConsumerImpl<T> newConsumer = new ConsumerImpl<>(client,
topicName, internalConfig,
- client.externalExecutorProvider().getExecutor(), 0, subFuture,
schema);
+ client.externalExecutorProvider().getExecutor(), 0, subFuture,
schema, interceptors);
consumers.putIfAbsent(newConsumer.getTopic(), newConsumer);
futureList = Collections.singletonList(subFuture);
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 3e159e8f48..12ecf2b0f2 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
@@ -53,8 +53,8 @@
private final TopicMetadata topicMetadata;
public PartitionedProducerImpl(PulsarClientImpl client, String topic,
ProducerConfigurationData conf, int numPartitions,
- CompletableFuture<Producer<T>> producerCreatedFuture, Schema<T>
schema) {
- super(client, topic, conf, producerCreatedFuture, schema);
+ CompletableFuture<Producer<T>> producerCreatedFuture, Schema<T>
schema, ProducerInterceptors<T> interceptors) {
+ super(client, topic, conf, producerCreatedFuture, schema,
interceptors);
this.producers = Lists.newArrayListWithCapacity(numPartitions);
this.topicMetadata = new TopicMetadataImpl(numPartitions);
this.routerPolicy = getMessageRouter();
@@ -111,7 +111,7 @@ private void start() {
for (int partitionIndex = 0; partitionIndex <
topicMetadata.numPartitions(); partitionIndex++) {
String partitionName =
TopicName.get(topic).getPartition(partitionIndex).toString();
ProducerImpl<T> producer = new ProducerImpl<>(client,
partitionName, conf, new CompletableFuture<>(),
- partitionIndex, schema);
+ partitionIndex, schema, interceptors);
producers.add(producer);
producer.producerCreatedFuture().handle((prod, createException) ->
{
if (createException != null) {
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java
index d0b0c606c5..3f6dfecdb3 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternMultiTopicsConsumerImpl.java
@@ -51,8 +51,8 @@ public PatternMultiTopicsConsumerImpl(Pattern topicsPattern,
ConsumerConfigurationData<T> conf,
ExecutorService listenerExecutor,
CompletableFuture<Consumer<T>>
subscribeFuture,
- Schema<T> schema) {
- super(client, conf, listenerExecutor, subscribeFuture, schema);
+ Schema<T> schema,
ConsumerInterceptors<T> interceptors) {
+ super(client, conf, listenerExecutor, subscribeFuture, schema,
interceptors);
this.topicsPattern = topicsPattern;
if (this.namespaceName == null) {
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 eb02b6bd8f..39e632c97a 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
@@ -36,13 +36,15 @@
protected final CompletableFuture<Producer<T>> producerCreatedFuture;
protected final ProducerConfigurationData conf;
protected final Schema<T> schema;
+ protected final ProducerInterceptors<T> interceptors;
protected ProducerBase(PulsarClientImpl client, String topic,
ProducerConfigurationData conf,
- CompletableFuture<Producer<T>> producerCreatedFuture, Schema<T>
schema) {
+ CompletableFuture<Producer<T>> producerCreatedFuture, Schema<T>
schema, ProducerInterceptors<T> interceptors) {
super(client, topic);
this.producerCreatedFuture = producerCreatedFuture;
this.conf = conf;
this.schema = schema;
+ this.interceptors = interceptors;
}
@Override
@@ -148,6 +150,20 @@ public ProducerConfigurationData getConfiguration() {
return producerCreatedFuture;
}
+ protected Message<T> beforeSend(Message<T> message) {
+ if (interceptors != null) {
+ return interceptors.beforeSend(this, message);
+ } else {
+ return message;
+ }
+ }
+
+ protected void onSendAcknowledgement(Message<T> message, MessageId msgId,
Throwable exception) {
+ if (interceptors != null) {
+ interceptors.onSendAcknowledgement(this, message, msgId,
exception);
+ }
+ }
+
@Override
public String toString() {
return "ProducerBase{" + "topic='" + topic + '\'' + '}';
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 ff391be3ee..834e0c082f 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
@@ -20,6 +20,9 @@
import com.google.common.annotations.VisibleForTesting;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -33,6 +36,7 @@
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
+import org.apache.pulsar.client.api.ProducerInterceptor;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
@@ -46,6 +50,7 @@
private final PulsarClientImpl client;
private ProducerConfigurationData conf;
private Schema<T> schema;
+ private List<ProducerInterceptor<T>> interceptorList;
@VisibleForTesting
public ProducerBuilderImpl(PulsarClientImpl client, Schema<T> schema) {
@@ -97,7 +102,9 @@ private ProducerBuilderImpl(PulsarClientImpl client,
ProducerConfigurationData c
.failedFuture(new IllegalArgumentException("Topic name
must be set on the producer builder"));
}
- return client.createProducerAsync(conf, schema);
+ return interceptorList == null || interceptorList.size() == 0 ?
+ client.createProducerAsync(conf, schema, null) :
+ client.createProducerAsync(conf, schema, new
ProducerInterceptors<>(interceptorList));
}
@Override
@@ -226,4 +233,13 @@ private ProducerBuilderImpl(PulsarClientImpl client,
ProducerConfigurationData c
conf.getProperties().putAll(properties);
return this;
}
+
+ @Override
+ public ProducerBuilder<T> intercept(ProducerInterceptor<T>...
interceptors) {
+ if (interceptorList == null) {
+ interceptorList = new ArrayList<>();
+ }
+ interceptorList.addAll(Arrays.asList(interceptors));
+ return this;
+ }
}
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 c3428e2644..9401e7cd20 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
@@ -121,8 +121,9 @@
.newUpdater(ProducerImpl.class, "msgIdGenerator");
public ProducerImpl(PulsarClientImpl client, String topic,
ProducerConfigurationData conf,
- CompletableFuture<Producer<T>> producerCreatedFuture,
int partitionIndex, Schema<T> schema) {
- super(client, topic, conf, producerCreatedFuture, schema);
+ CompletableFuture<Producer<T>> producerCreatedFuture,
int partitionIndex, Schema<T> schema,
+ ProducerInterceptors<T> interceptors) {
+ super(client, topic, conf, producerCreatedFuture, schema,
interceptors);
this.producerId = client.newProducerId();
this.producerName = conf.getProducerName();
this.partitionIndex = partitionIndex;
@@ -205,9 +206,16 @@ public long getLastSequenceId() {
@Override
CompletableFuture<MessageId> internalSendAsync(Message<T> message) {
+
CompletableFuture<MessageId> future = new CompletableFuture<>();
- sendAsync(message, new SendCallback() {
+ MessageImpl<T> interceptorMessage = (MessageImpl<T>)
beforeSend(message);
+ //Retain the buffer used by interceptors callback to get message.
Buffer will release after complete interceptors.
+ interceptorMessage.getDataBuffer().retain();
+ if (interceptors != null) {
+ interceptorMessage.getProperties();
+ }
+ sendAsync(interceptorMessage, new SendCallback() {
SendCallback nextCallback = null;
MessageImpl<?> nextMsg = null;
long createdAt = System.nanoTime();
@@ -229,25 +237,40 @@ public SendCallback getNextSendCallback() {
@Override
public void sendComplete(Exception e) {
- if (e != null) {
- stats.incrementSendFailed();
- future.completeExceptionally(e);
- } else {
- future.complete(message.getMessageId());
- stats.incrementNumAcksReceived(System.nanoTime() -
createdAt);
- }
- while (nextCallback != null) {
- SendCallback sendCallback = nextCallback;
- MessageImpl<?> msg = nextMsg;
+ try {
if (e != null) {
stats.incrementSendFailed();
- sendCallback.getFuture().completeExceptionally(e);
+ onSendAcknowledgement(interceptorMessage, null, e);
+ future.completeExceptionally(e);
} else {
- sendCallback.getFuture().complete(msg.getMessageId());
+ onSendAcknowledgement(interceptorMessage,
interceptorMessage.getMessageId(), null);
+ future.complete(interceptorMessage.getMessageId());
stats.incrementNumAcksReceived(System.nanoTime() -
createdAt);
}
- nextMsg = nextCallback.getNextMessage();
- nextCallback = nextCallback.getNextSendCallback();
+ } finally {
+ interceptorMessage.getDataBuffer().release();
+ }
+
+ while (nextCallback != null) {
+ SendCallback sendCallback = nextCallback;
+ MessageImpl<?> msg = nextMsg;
+ //Retain the buffer used by interceptors callback to get
message. Buffer will release after complete interceptors.
+ try {
+ msg.getDataBuffer().retain();
+ if (e != null) {
+ stats.incrementSendFailed();
+ onSendAcknowledgement((Message<T>) msg, null, e);
+ sendCallback.getFuture().completeExceptionally(e);
+ } else {
+ onSendAcknowledgement((Message<T>) msg,
msg.getMessageId(), null);
+
sendCallback.getFuture().complete(msg.getMessageId());
+ stats.incrementNumAcksReceived(System.nanoTime() -
createdAt);
+ }
+ nextMsg = nextCallback.getNextMessage();
+ nextCallback = nextCallback.getNextSendCallback();
+ } finally {
+ msg.getDataBuffer().release();
+ }
}
}
@@ -292,14 +315,18 @@ public void sendAsync(Message<T> message, SendCallback
callback) {
String compressedStr = (!isBatchMessagingEnabled() &&
conf.getCompressionType() != CompressionType.NONE)
? "Compressed"
: "";
- callback.sendComplete(new
PulsarClientException.InvalidMessageException(
- format("%s Message payload size %d cannot exceed %d
bytes", compressedStr, compressedSize,
- PulsarDecoder.MaxMessageSize)));
+ PulsarClientException.InvalidMessageException
invalidMessageException =
+ new PulsarClientException.InvalidMessageException(
+ format("%s Message payload size %d cannot exceed
%d bytes", compressedStr, compressedSize,
+ PulsarDecoder.MaxMessageSize));
+ callback.sendComplete(invalidMessageException);
return;
}
if (!msg.isReplicated() && msgMetadataBuilder.hasProducerName()) {
- callback.sendComplete(new
PulsarClientException.InvalidMessageException("Cannot re-use the same
message"));
+ PulsarClientException.InvalidMessageException
invalidMessageException =
+ new PulsarClientException.InvalidMessageException("Cannot
re-use the same message");
+ callback.sendComplete(invalidMessageException);
compressedPayload.release();
return;
}
@@ -463,8 +490,7 @@ private boolean canEnqueueRequest(SendCallback callback) {
semaphore.acquire();
} else {
if (!semaphore.tryAcquire()) {
- callback.sendComplete(
- new
PulsarClientException.ProducerQueueIsFullError("Producer send queue is full"));
+ callback.sendComplete(new
PulsarClientException.ProducerQueueIsFullError("Producer send queue is full"));
return false;
}
}
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
new file mode 100644
index 0000000000..59d1ad3183
--- /dev/null
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerInterceptors.java
@@ -0,0 +1,110 @@
+/**
+ * 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 org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerInterceptor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * A container that holds the list{@link
org.apache.pulsar.client.api.ProducerInterceptor}
+ * and wraps calls to the chain of custom interceptors.
+ */
+public class ProducerInterceptors<T> implements Closeable {
+
+ private static final Logger log =
LoggerFactory.getLogger(ProducerInterceptors.class);
+
+ private final List<ProducerInterceptor<T>> interceptors;
+
+ public ProducerInterceptors(List<ProducerInterceptor<T>> interceptors) {
+ this.interceptors = interceptors;
+ }
+
+ /**
+ * This is called when client sends message to pulsar broker, before key
and value gets serialized.
+ * The method calls {@link
ProducerInterceptor#beforeSend(Producer,Message)} method. Message returned from
+ * first interceptor's beforeSend() is passed to the second interceptor
beforeSend(), and so on in the
+ * interceptor chain. The message returned from the last interceptor is
returned from this method.
+ *
+ * This method does not throw exceptions. Exceptions thrown by any
interceptor methods are caught and ignored.
+ * If a interceptor in the middle of the chain, that normally modifies the
message, throws an exception,
+ * the next interceptor in the chain will be called with a message
returned by the previous interceptor that did
+ * not throw an exception.
+ *
+ * @param producer the producer which contains the interceptor.
+ * @param message the message from client
+ * @return the message to send to topic/partition
+ */
+ public Message<T> beforeSend(Producer<T> producer, Message<T> message) {
+ Message<T> interceptorMessage = message;
+ for (int i = 0; i < interceptors.size(); i++) {
+ try {
+ interceptorMessage = interceptors.get(i).beforeSend(producer,
interceptorMessage);
+ } catch (Exception e) {
+ if (message != null && producer != null) {
+ log.warn("Error executing interceptor beforeSend callback
for messageId: {}, topicName:{} ", message.getMessageId(), producer.getTopic(),
e);
+ } else {
+ log.warn("Error Error executing interceptor beforeSend
callback ", e);
+ }
+ }
+ }
+ return interceptorMessage;
+ }
+
+ /**
+ * This method is called when the message send to the broker has been
acknowledged, or when sending the record fails
+ * before it gets send to the broker.
+ * This method calls {@link
ProducerInterceptor#onSendAcknowledgement(Producer, Message, MessageId,
Throwable)} method for
+ * each interceptor.
+ *
+ * This method does not throw exceptions. Exceptions thrown by any of
interceptor methods are caught and ignored.
+ *
+ * @param producer the producer which contains the interceptor.
+ * @param message The message returned from the last interceptor is
returned from {@link ProducerInterceptor#beforeSend(Producer, Message)}
+ * @param msgId The message id that broker returned. Null if has error
occurred.
+ * @param exception The exception thrown during processing of this
message. Null if no error occurred.
+ */
+ public void onSendAcknowledgement(Producer<T> producer, Message<T>
message, MessageId msgId, Throwable exception) {
+ for (int i = 0; i < interceptors.size(); i++) {
+ try {
+ interceptors.get(i).onSendAcknowledgement(producer, message,
msgId, exception);
+ } catch (Exception e) {
+ log.warn("Error executing interceptor onSendAcknowledgement
callback ", e);
+ }
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ for (int i = 0; i < interceptors.size(); i++) {
+ try {
+ interceptors.get(i).close();
+ } catch (Exception e) {
+ log.error("Fail to close producer interceptor ", e);
+ }
+ }
+ }
+}
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 4013068246..b87b9bd913 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
@@ -238,10 +238,15 @@ public ClientConfigurationData getConfiguration() {
}
public CompletableFuture<Producer<byte[]>>
createProducerAsync(ProducerConfigurationData conf) {
- return createProducerAsync(conf, Schema.BYTES);
+ return createProducerAsync(conf, Schema.BYTES, null);
}
- public <T> CompletableFuture<Producer<T>>
createProducerAsync(ProducerConfigurationData conf, Schema<T> schema) {
+ public <T> CompletableFuture<Producer<T>>
createProducerAsync(ProducerConfigurationData conf, Schema<T> schema) {
+ return createProducerAsync(conf, schema, null);
+ }
+
+ public <T> CompletableFuture<Producer<T>>
createProducerAsync(ProducerConfigurationData conf, Schema<T> schema,
+ ProducerInterceptors<T> interceptors) {
if (conf == null) {
return FutureUtil.failedFuture(
new
PulsarClientException.InvalidConfigurationException("Producer configuration
undefined"));
@@ -273,9 +278,9 @@ public ClientConfigurationData getConfiguration() {
ProducerBase<T> producer;
if (metadata.partitions > 1) {
producer = new
PartitionedProducerImpl<>(PulsarClientImpl.this, topic, conf,
metadata.partitions,
- producerCreatedFuture, schema);
+ producerCreatedFuture, schema, interceptors);
} else {
- producer = new ProducerImpl<>(PulsarClientImpl.this, topic,
conf, producerCreatedFuture, -1, schema);
+ producer = new ProducerImpl<>(PulsarClientImpl.this, topic,
conf, producerCreatedFuture, -1, schema, interceptors);
}
synchronized (producers) {
@@ -336,10 +341,10 @@ public ClientConfigurationData getConfiguration() {
}
public CompletableFuture<Consumer<byte[]>>
subscribeAsync(ConsumerConfigurationData<byte[]> conf) {
- return subscribeAsync(conf, Schema.BYTES);
+ return subscribeAsync(conf, Schema.BYTES, null);
}
- public <T> CompletableFuture<Consumer<T>>
subscribeAsync(ConsumerConfigurationData<T> conf, Schema<T> schema) {
+ public <T> CompletableFuture<Consumer<T>>
subscribeAsync(ConsumerConfigurationData<T> conf, Schema<T> schema,
ConsumerInterceptors<T> interceptors) {
if (state.get() != State.Open) {
return FutureUtil.failedFuture(new
PulsarClientException.AlreadyClosedException("Client already closed"));
}
@@ -377,15 +382,15 @@ public ClientConfigurationData getConfiguration() {
return FutureUtil
.failedFuture(new IllegalArgumentException("Topic names
list must be null when use topicsPattern"));
}
- return patternTopicSubscribeAsync(conf, schema);
+ return patternTopicSubscribeAsync(conf, schema, interceptors);
} else if (conf.getTopicNames().size() == 1) {
- return singleTopicSubscribeAsync(conf, schema);
+ return singleTopicSubscribeAsync(conf, schema, interceptors);
} else {
- return multiTopicSubscribeAsync(conf, schema);
+ return multiTopicSubscribeAsync(conf, schema, interceptors);
}
}
- private <T> CompletableFuture<Consumer<T>>
singleTopicSubscribeAsync(ConsumerConfigurationData<T> conf, Schema<T> schema) {
+ private <T> CompletableFuture<Consumer<T>>
singleTopicSubscribeAsync(ConsumerConfigurationData<T> conf, Schema<T> schema,
ConsumerInterceptors<T> interceptors) {
if (schema instanceof AutoSchema) {
AutoSchema autoSchema = (AutoSchema) schema;
return lookup.getSchema(TopicName.get(conf.getSingleTopic()))
@@ -395,20 +400,20 @@ public ClientConfigurationData getConfiguration() {
log.info("Auto detected schema for topic {} : {}",
conf.getSingleTopic(), new
String(schemaInfoOptional.get().getSchema(), UTF_8));
autoSchema.setSchema(genericSchema);
- return doSingleTopicSubscribeAsync(conf, schema);
+ return doSingleTopicSubscribeAsync(conf, schema,
interceptors);
} else {
return FutureUtil.failedFuture(
new
PulsarClientException.LookupException("Currently schema detection only works
for topics with avro schemas"));
}
});
} else {
- return doSingleTopicSubscribeAsync(conf, schema);
+ return doSingleTopicSubscribeAsync(conf, schema, interceptors);
}
}
- private <T> CompletableFuture<Consumer<T>>
doSingleTopicSubscribeAsync(ConsumerConfigurationData<T> conf, Schema<T>
schema) {
+ private <T> CompletableFuture<Consumer<T>>
doSingleTopicSubscribeAsync(ConsumerConfigurationData<T> conf, Schema<T>
schema, ConsumerInterceptors<T> interceptors) {
CompletableFuture<Consumer<T>> consumerSubscribedFuture = new
CompletableFuture<>();
String topic = conf.getSingleTopic();
@@ -423,10 +428,10 @@ public ClientConfigurationData getConfiguration() {
ExecutorService listenerThread =
externalExecutorProvider.getExecutor();
if (metadata.partitions > 1) {
consumer =
MultiTopicsConsumerImpl.createPartitionedConsumer(PulsarClientImpl.this, conf,
- listenerThread, consumerSubscribedFuture,
metadata.partitions, schema);
+ listenerThread, consumerSubscribedFuture,
metadata.partitions, schema, interceptors);
} else {
consumer = new ConsumerImpl<>(PulsarClientImpl.this, topic,
conf, listenerThread, -1,
- consumerSubscribedFuture, schema);
+ consumerSubscribedFuture, schema, interceptors);
}
synchronized (consumers) {
@@ -441,11 +446,11 @@ public ClientConfigurationData getConfiguration() {
return consumerSubscribedFuture;
}
- private <T> CompletableFuture<Consumer<T>>
multiTopicSubscribeAsync(ConsumerConfigurationData<T> conf, Schema<T> schema) {
+ private <T> CompletableFuture<Consumer<T>>
multiTopicSubscribeAsync(ConsumerConfigurationData<T> conf, Schema<T> schema,
ConsumerInterceptors<T> interceptors) {
CompletableFuture<Consumer<T>> consumerSubscribedFuture = new
CompletableFuture<>();
ConsumerBase<T> consumer = new
MultiTopicsConsumerImpl<>(PulsarClientImpl.this, conf,
- externalExecutorProvider.getExecutor(),
consumerSubscribedFuture, schema);
+ externalExecutorProvider.getExecutor(),
consumerSubscribedFuture, schema, interceptors);
synchronized (consumers) {
consumers.put(consumer, Boolean.TRUE);
@@ -455,10 +460,10 @@ public ClientConfigurationData getConfiguration() {
}
public CompletableFuture<Consumer<byte[]>>
patternTopicSubscribeAsync(ConsumerConfigurationData<byte[]> conf) {
- return patternTopicSubscribeAsync(conf, Schema.BYTES);
+ return patternTopicSubscribeAsync(conf, Schema.BYTES, null);
}
- private <T> CompletableFuture<Consumer<T>>
patternTopicSubscribeAsync(ConsumerConfigurationData<T> conf, Schema<T> schema)
{
+ private <T> CompletableFuture<Consumer<T>>
patternTopicSubscribeAsync(ConsumerConfigurationData<T> conf, Schema<T> schema,
ConsumerInterceptors interceptors) {
String regex = conf.getTopicsPattern().pattern();
TopicName destination = TopicName.get(regex);
NamespaceName namespaceName = destination.getNamespaceObject();
@@ -479,7 +484,7 @@ public ClientConfigurationData getConfiguration() {
conf,
externalExecutorProvider.getExecutor(),
consumerSubscribedFuture,
- schema);
+ schema, interceptors);
synchronized (consumers) {
consumers.put(consumer, Boolean.TRUE);
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
index aafd12549f..388e5c487d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
@@ -85,7 +85,7 @@ public void reachedEndOfTopic(Consumer<T> consumer) {
final int partitionIdx =
TopicName.getPartitionIndex(readerConfiguration.getTopicName());
consumer = new ConsumerImpl<>(client,
readerConfiguration.getTopicName(), consumerConfiguration, listenerExecutor,
- partitionIdx, consumerFuture, SubscriptionMode.NonDurable,
readerConfiguration.getStartMessageId(), schema);
+ partitionIdx, consumerFuture, SubscriptionMode.NonDurable,
readerConfiguration.getStartMessageId(), schema, null);
}
@Override
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageImpl.java
index eae02b08c0..230a022c61 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageImpl.java
@@ -127,4 +127,8 @@ public T getValue() {
public Optional<EncryptionContext> getEncryptionCtx() {
return msg.getEncryptionCtx();
}
+
+ public Message<T> getMessage() {
+ return msg;
+ }
}
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 4380c63ff7..f5108fc6c1 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
@@ -64,7 +64,7 @@ public void setup() {
logger = mock(Logger.class);
client = mock(PulsarClientImpl.class);
when(client.newProducer()).thenReturn(new ProducerBuilderImpl(client,
Schema.BYTES));
-
when(client.createProducerAsync(Matchers.any(ProducerConfigurationData.class),
Matchers.any(Schema.class)))
+
when(client.createProducerAsync(Matchers.any(ProducerConfigurationData.class),
Matchers.any(Schema.class), eq(null)))
.thenReturn(CompletableFuture.completedFuture(producer));
when(client.getSchema(anyString())).thenReturn(CompletableFuture.completedFuture(Optional.empty()));
when(producer.sendAsync(anyString())).thenReturn(CompletableFuture.completedFuture(null));
diff --git
a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/producers/MultiConsumersOneOutputTopicProducersTest.java
b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/producers/MultiConsumersOneOutputTopicProducersTest.java
index 89efcf594d..399a4c701b 100644
---
a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/producers/MultiConsumersOneOutputTopicProducersTest.java
+++
b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/producers/MultiConsumersOneOutputTopicProducersTest.java
@@ -21,7 +21,8 @@
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.any;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
@@ -40,6 +41,7 @@
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
+import org.apache.pulsar.client.api.ProducerInterceptor;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
@@ -190,6 +192,11 @@
public ProducerBuilder<byte[]> properties(Map<String, String>
properties) {
return this;
}
+
+ @Override
+ public ProducerBuilder<byte[]>
intercept(ProducerInterceptor<byte[]>... interceptors) {
+ return null;
+ }
}
@BeforeMethod
@@ -197,7 +204,7 @@ public void setup() throws Exception {
this.mockClient = mock(PulsarClient.class);
when(mockClient.newProducer(any(Schema.class)))
- .thenReturn(new MockProducerBuilder());
+ .thenReturn(new MockProducerBuilder());
producers = new
MultiConsumersOneOuputTopicProducers<byte[]>(mockClient, TEST_OUTPUT_TOPIC,
Schema.BYTES, "test");
producers.initialize();
@@ -206,12 +213,12 @@ public void setup() throws Exception {
private Producer<byte[]> createMockProducer(String topic) {
Producer<byte[]> producer = mock(Producer.class);
when(producer.closeAsync())
- .thenAnswer(invocationOnMock -> {
- synchronized (mockProducers) {
- mockProducers.remove(topic);
- }
- return FutureUtils.Void();
- });
+ .thenAnswer(invocationOnMock -> {
+ synchronized (mockProducers) {
+ mockProducers.remove(topic);
+ }
+ return FutureUtils.Void();
+ });
return producer;
}
@@ -224,13 +231,13 @@ public void testGetCloseProducer() throws Exception {
assertSame(mockProducers.get(producerName), producer);
verify(mockClient, times(1))
- .newProducer(Schema.BYTES);
+ .newProducer(Schema.BYTES);
assertTrue(producers.getProducers().containsKey(producerName));
// second get will not create a new producer
assertSame(mockProducers.get(producerName), producer);
verify(mockClient, times(1))
- .newProducer(Schema.BYTES);
+ .newProducer(Schema.BYTES);
assertTrue(producers.getProducers().containsKey(producerName));
// close
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
