You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2020/12/21 11:55:22 UTC
[pulsar] 01/03: Intercept beforeSendMessage calls (#8932)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.6
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d3465529d33f0bf35fe837444e2216085eecbc5f
Author: Sijie Guo <si...@apache.org>
AuthorDate: Mon Dec 14 16:26:16 2020 -0800
Intercept beforeSendMessage calls (#8932)
*Motivation*
Currently the message metadata headers are deserialized for filtering
out the entries to dispatch. Add a `beforeSendMessage` method to
intercept entries before sending them to consumers
(cherry picked from commit f76655a9d23d1130cb89e4bf5431722f470efbf4)
---
.../pulsar/broker/intercept/BrokerInterceptor.java | 26 +++++++++++++++----
.../BrokerInterceptorWithClassLoader.java | 12 +++++++++
.../broker/intercept/BrokerInterceptors.java | 17 ++++++++++++
.../broker/service/AbstractBaseDispatcher.java | 19 ++++++++++----
.../apache/pulsar/broker/service/Subscription.java | 3 +++
.../nonpersistent/NonPersistentSubscription.java | 6 +++++
.../service/persistent/PersistentSubscription.java | 6 +++++
.../broker/intercept/BrokerInterceptorTest.java | 30 +++++++++++++++++++++-
.../broker/intercept/CounterBrokerInterceptor.java | 18 +++++++++++++
9 files changed, 126 insertions(+), 11 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptor.java
index 164f757..f1bbb2e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptor.java
@@ -19,16 +19,18 @@
package org.apache.pulsar.broker.intercept;
import com.google.common.annotations.Beta;
+import java.io.IOException;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.common.api.proto.PulsarApi.BaseCommand;
+import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.intercept.InterceptException;
-import javax.servlet.ServletException;
-import javax.servlet.ServletRequest;
-import javax.servlet.ServletResponse;
-import java.io.IOException;
-
/**
* A plugin interface that allows you to intercept the
* client requests to the Pulsar brokers.
@@ -40,6 +42,20 @@ import java.io.IOException;
public interface BrokerInterceptor extends AutoCloseable {
/**
+ * Intercept messages before sending them to the consumers.
+ *
+ * @param subscription pulsar subscription
+ * @param entry entry
+ * @param ackSet entry ack bitset. it is either <tt>null</tt> or an array of long-based bitsets.
+ * @param msgMetadata message metadata. The message metadata will be recycled after this call.
+ */
+ default void beforeSendMessage(Subscription subscription,
+ Entry entry,
+ long[] ackSet,
+ MessageMetadata msgMetadata) {
+ }
+
+ /**
* Called by the broker while new command incoming.
*/
void onPulsarCommand(BaseCommand command, ServerCnx cnx) throws InterceptException;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoader.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoader.java
index 00c960c..68710cd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoader.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptorWithClassLoader.java
@@ -21,9 +21,12 @@ package org.apache.pulsar.broker.intercept;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.common.api.proto.PulsarApi.BaseCommand;
+import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.intercept.InterceptException;
import org.apache.pulsar.common.nar.NarClassLoader;
@@ -44,6 +47,15 @@ public class BrokerInterceptorWithClassLoader implements BrokerInterceptor {
private final NarClassLoader classLoader;
@Override
+ public void beforeSendMessage(Subscription subscription,
+ Entry entry,
+ long[] ackSet,
+ MessageMetadata msgMetadata) {
+ this.interceptor.beforeSendMessage(
+ subscription, entry, ackSet, msgMetadata);
+ }
+
+ @Override
public void onPulsarCommand(BaseCommand command, ServerCnx cnx) throws InterceptException {
this.interceptor.onPulsarCommand(command, cnx);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java
index 6f0b7b8..77cca6d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java
@@ -20,10 +20,13 @@ package org.apache.pulsar.broker.intercept;
import com.google.common.collect.ImmutableMap;
import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.common.api.proto.PulsarApi.BaseCommand;
+import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.intercept.InterceptException;
import javax.servlet.ServletException;
@@ -86,6 +89,20 @@ public class BrokerInterceptors implements BrokerInterceptor {
}
@Override
+ public void beforeSendMessage(Subscription subscription,
+ Entry entry,
+ long[] ackSet,
+ MessageMetadata msgMetadata) {
+ for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ value.beforeSendMessage(
+ subscription,
+ entry,
+ ackSet,
+ msgMetadata);
+ }
+ }
+
+ @Override
public void onPulsarCommand(BaseCommand command, ServerCnx cnx) throws InterceptException {
for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
value.onPulsarCommand(command, cnx);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
index 7cf9793..29ac54b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
@@ -21,7 +21,6 @@ package org.apache.pulsar.broker.service;
import io.netty.buffer.ByteBuf;
-import java.io.IOException;
import java.util.Collections;
import java.util.List;
@@ -30,11 +29,9 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
-import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.common.api.proto.PulsarApi;
-import org.apache.pulsar.common.compression.CompressionCodec;
-import org.apache.pulsar.common.compression.CompressionCodecProvider;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.protocol.Markers;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
@@ -113,14 +110,26 @@ public abstract class AbstractBaseDispatcher implements Dispatcher {
totalBytes += metadataAndPayload.readableBytes();
totalChunkedMessages += msgMetadata.hasChunkId() ? 1: 0;
batchSizes.setBatchSize(i, batchSize);
+ long[] ackSet = null;
if (indexesAcks != null && cursor != null) {
- long[] ackSet = cursor.getDeletedBatchIndexesAsLongArray(PositionImpl.get(entry.getLedgerId(), entry.getEntryId()));
+ ackSet = cursor.getDeletedBatchIndexesAsLongArray(PositionImpl.get(entry.getLedgerId(), entry.getEntryId()));
if (ackSet != null) {
indexesAcks.setIndexesAcks(i, Pair.of(batchSize, ackSet));
} else {
indexesAcks.setIndexesAcks(i,null);
}
}
+
+ BrokerInterceptor interceptor = subscription.interceptor();
+ if (null != interceptor) {
+ interceptor.beforeSendMessage(
+ subscription,
+ entry,
+ ackSet,
+ msgMetadata
+ );
+ }
+
} finally {
msgMetadata.recycle();
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java
index 85b5415..88ac442 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java
@@ -25,12 +25,15 @@ import java.util.concurrent.CompletableFuture;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType;
import org.apache.pulsar.common.api.proto.PulsarMarkers.ReplicatedSubscriptionsSnapshot;
public interface Subscription {
+ BrokerInterceptor interceptor();
+
Topic getTopic();
String getName();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
index 6421bde..90be218 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
@@ -29,6 +29,7 @@ import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException;
import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException;
@@ -77,6 +78,11 @@ public class NonPersistentSubscription implements Subscription {
}
@Override
+ public BrokerInterceptor interceptor() {
+ return this.topic.getBrokerService().getInterceptor();
+ }
+
+ @Override
public String getName() {
return this.subName;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index 51b1e99..9958e7d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -50,6 +50,7 @@ import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.bookkeeper.util.collections.ConcurrentLongLongPairHashMap;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException;
import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException;
@@ -153,6 +154,11 @@ public class PersistentSubscription implements Subscription {
}
@Override
+ public BrokerInterceptor interceptor() {
+ return topic.getBrokerService().getInterceptor();
+ }
+
+ @Override
public String getName() {
return this.subName;
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java
index 5ff73fc..1cd10a5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java
@@ -18,8 +18,11 @@
*/
package org.apache.pulsar.broker.intercept;
-import org.apache.pulsar.broker.ServiceConfiguration;
+import lombok.Cleanup;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
@@ -36,6 +39,7 @@ import static org.mockito.ArgumentMatchers.same;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.testng.Assert.assertEquals;
public class BrokerInterceptorTest extends ProducerConsumerBase {
@@ -109,4 +113,28 @@ public class BrokerInterceptorTest extends ProducerConsumerBase {
// CONNECT and PRODUCER
Assert.assertTrue(((CounterBrokerInterceptor)listener).getCount() >= 2);
}
+
+ @Test
+ public void testBeforeSendMessage() throws PulsarClientException {
+ BrokerInterceptor listener = pulsar.getBrokerInterceptor();
+ Assert.assertTrue(listener instanceof CounterBrokerInterceptor);
+
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic("test-before-send-message")
+ .create();
+
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic("test-before-send-message")
+ .subscriptionName("test")
+ .subscribe();
+
+ producer.send("hello world");
+
+ Message<String> msg = consumer.receive();
+
+ assertEquals(msg.getValue(), "hello world");
+
+ Assert.assertTrue(((CounterBrokerInterceptor) listener).getBeforeSendCount() == 1);
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java
index 4436850..6915101 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/CounterBrokerInterceptor.java
@@ -19,9 +19,12 @@
package org.apache.pulsar.broker.intercept;
import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.common.api.proto.PulsarApi;
+import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.intercept.InterceptException;
import javax.servlet.FilterChain;
@@ -34,9 +37,20 @@ import java.io.IOException;
@Slf4j
public class CounterBrokerInterceptor implements BrokerInterceptor {
+ int beforeSendCount = 0;
int count = 0;
@Override
+ public void beforeSendMessage(Subscription subscription,
+ Entry entry,
+ long[] ackSet,
+ MessageMetadata msgMetadata) {
+ log.info("Send message to topic {}, subscription {}",
+ subscription.getTopic(), subscription.getName());
+ beforeSendCount++;
+ }
+
+ @Override
public void onPulsarCommand(PulsarApi.BaseCommand command, ServerCnx cnx) throws InterceptException {
log.info("[{}] On [{}] Pulsar command", count, command.getType().name());
count ++;
@@ -72,4 +86,8 @@ public class CounterBrokerInterceptor implements BrokerInterceptor {
public int getCount() {
return count;
}
+
+ public int getBeforeSendCount() {
+ return beforeSendCount;
+ }
}