You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by bo...@apache.org on 2022/11/30 09:12:54 UTC
[pulsar] 01/02: [pulsar-broker] Broker extensions to allow operators of enterprise wide cluster better control and flexibility (#12536)
This is an automated email from the ASF dual-hosted git repository.
bogong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c7124bb1325126b1151ebc655cba0423f4adf2d7
Author: madhavan-narayanan <92...@users.noreply.github.com>
AuthorDate: Tue Dec 14 12:19:07 2021 +0530
[pulsar-broker] Broker extensions to allow operators of enterprise wide cluster better control and flexibility (#12536)
### Motivation
Operators of enterprise Pulsar cluster(s) would need the flexibility and control to intercept broker events (including ledger writes/reads) for template validations, observability and access control. This changeset contains enhancements to existing interceptor mechanism to support this
### Modifications
- Enhanced org.apache.pulsar.broker.intercept.BrokerInterceptor interface to include additional events for tracing
- Created a new interface org.apache.pulsar.common.intercept.MessagePayloadProcessor to allow interception of ledger write/read operations
- Enhanced PulsarAdmin to give operators a control in managing super-users
### Verifying this change
- [x ] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
- *Added new test cases to MangedLedgerInterceptorImplTest.java and BrokerInterceptorTest.java *
(cherry picked from commit 03bbc8e67ad429a774936d26ab2f4d40c1ea4999)
---
.../bookkeeper/mledger/impl/EntryCacheImpl.java | 7 +-
.../bookkeeper/mledger/impl/EntryCacheManager.java | 29 +++++++-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 6 +-
.../apache/bookkeeper/mledger/impl/OpAddEntry.java | 20 +++++
.../intercept/ManagedLedgerInterceptor.java | 35 +++++++++
.../apache/pulsar/broker/ServiceConfiguration.java | 6 ++
.../pulsar/broker/intercept/BrokerInterceptor.java | 65 +++++++++++++++++
.../BrokerInterceptorWithClassLoader.java | 43 +++++++++++
.../broker/intercept/BrokerInterceptors.java | 74 +++++++++++++++++++
.../intercept/ManagedLedgerInterceptorImpl.java | 63 +++++++++++++++-
.../pulsar/broker/service/BrokerService.java | 15 +++-
.../org/apache/pulsar/broker/service/Consumer.java | 4 +
.../org/apache/pulsar/broker/service/Producer.java | 36 +++++++++
.../apache/pulsar/broker/service/ServerCnx.java | 27 ++++++-
.../org/apache/pulsar/broker/service/Topic.java | 7 ++
.../broker/intercept/BrokerInterceptorTest.java | 32 ++++++++
.../broker/intercept/CounterBrokerInterceptor.java | 85 ++++++++++++++++++++++
.../intercept/MangedLedgerInterceptorImplTest.java | 75 ++++++++++++++++++-
.../common/intercept/BrokerEntryMetadataUtils.java | 27 ++++++-
.../intercept/ManagedLedgerPayloadProcessor.java | 62 ++++++++++++++++
20 files changed, 702 insertions(+), 16 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java
index 085923e25cc..0851d942bd1 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java
@@ -38,6 +38,7 @@ import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
import org.apache.bookkeeper.mledger.util.RangeCache;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
@@ -50,6 +51,7 @@ public class EntryCacheImpl implements EntryCache {
private final EntryCacheManager manager;
private final ManagedLedgerImpl ml;
+ private ManagedLedgerInterceptor interceptor;
private final RangeCache<PositionImpl, EntryImpl> entries;
private final boolean copyEntries;
@@ -58,6 +60,7 @@ public class EntryCacheImpl implements EntryCache {
public EntryCacheImpl(EntryCacheManager manager, ManagedLedgerImpl ml, boolean copyEntries) {
this.manager = manager;
this.ml = ml;
+ this.interceptor = ml.getManagedLedgerInterceptor();
this.entries = new RangeCache<>(EntryImpl::getLength, EntryImpl::getTimestamp);
this.copyEntries = copyEntries;
@@ -215,7 +218,7 @@ public class EntryCacheImpl implements EntryCache {
Iterator<LedgerEntry> iterator = ledgerEntries.iterator();
if (iterator.hasNext()) {
LedgerEntry ledgerEntry = iterator.next();
- EntryImpl returnEntry = EntryImpl.create(ledgerEntry);
+ EntryImpl returnEntry = EntryCacheManager.create(ledgerEntry, interceptor);
manager.mlFactoryMBean.recordCacheMiss(1, returnEntry.getLength());
ml.mbean.addReadEntriesSample(1, returnEntry.getLength());
@@ -301,7 +304,7 @@ public class EntryCacheImpl implements EntryCache {
final List<EntryImpl> entriesToReturn
= Lists.newArrayListWithExpectedSize(entriesToRead);
for (LedgerEntry e : ledgerEntries) {
- EntryImpl entry = EntryImpl.create(e);
+ EntryImpl entry = EntryCacheManager.create(e, interceptor);
entriesToReturn.add(entry);
totalSize += entry.getLength();
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java
index 1c0c288e878..9a1d6317e4d 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java
@@ -33,10 +33,12 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.bookkeeper.client.api.LedgerEntry;
import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -154,9 +156,11 @@ public class EntryCacheManager {
protected class EntryCacheDisabled implements EntryCache {
private final ManagedLedgerImpl ml;
+ private final ManagedLedgerInterceptor interceptor;
public EntryCacheDisabled(ManagedLedgerImpl ml) {
this.ml = ml;
+ this.interceptor = ml.getManagedLedgerInterceptor();
}
@Override
@@ -200,7 +204,7 @@ public class EntryCacheManager {
try {
for (LedgerEntry e : ledgerEntries) {
// Insert the entries at the end of the list (they will be unsorted for now)
- EntryImpl entry = EntryImpl.create(e);
+ EntryImpl entry = create(e, interceptor);
entries.add(entry);
totalSize += entry.getLength();
}
@@ -232,7 +236,7 @@ public class EntryCacheManager {
Iterator<LedgerEntry> iterator = ledgerEntries.iterator();
if (iterator.hasNext()) {
LedgerEntry ledgerEntry = iterator.next();
- EntryImpl returnEntry = EntryImpl.create(ledgerEntry);
+ EntryImpl returnEntry = create(ledgerEntry, interceptor);
mlFactoryMBean.recordCacheMiss(1, returnEntry.getLength());
ml.getMBean().addReadEntriesSample(1, returnEntry.getLength());
@@ -262,5 +266,26 @@ public class EntryCacheManager {
return EntryImpl.create(ledgerId, entryId, data);
}
+ public static EntryImpl create(LedgerEntry ledgerEntry, ManagedLedgerInterceptor interceptor) {
+ ManagedLedgerInterceptor.PayloadProcessorHandle processorHandle = null;
+ if (interceptor != null) {
+ ByteBuf duplicateBuffer = ledgerEntry.getEntryBuffer().retainedDuplicate();
+ processorHandle = interceptor
+ .processPayloadBeforeEntryCache(duplicateBuffer);
+ if (processorHandle != null) {
+ ledgerEntry = LedgerEntryImpl.create(ledgerEntry.getLedgerId(),ledgerEntry.getEntryId(),
+ ledgerEntry.getLength(),processorHandle.getProcessedPayload());
+ } else {
+ duplicateBuffer.release();
+ }
+ }
+ EntryImpl returnEntry = EntryImpl.create(ledgerEntry);
+ if (processorHandle != null) {
+ processorHandle.release();
+ ledgerEntry.close();
+ }
+ return returnEntry;
+ }
+
private static final Logger log = LoggerFactory.getLogger(EntryCacheManager.class);
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index ce34d092716..27b0141e129 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -310,6 +310,9 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
STATE_UPDATER.set(this, State.None);
this.ledgersStat = null;
this.mbean = new ManagedLedgerMBeanImpl(this);
+ if (config.getManagedLedgerInterceptor() != null) {
+ this.managedLedgerInterceptor = config.getManagedLedgerInterceptor();
+ }
this.entryCache = factory.getEntryCacheManager().getEntryCache(this);
this.waitingCursors = Queues.newConcurrentLinkedQueue();
this.waitingEntryCallBacks = Queues.newConcurrentLinkedQueue();
@@ -320,9 +323,6 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
this.maximumRolloverTimeMs = getMaximumRolloverTimeMs(config);
this.mlOwnershipChecker = mlOwnershipChecker;
this.propertiesMap = Maps.newHashMap();
- if (config.getManagedLedgerInterceptor() != null) {
- this.managedLedgerInterceptor = config.getManagedLedgerInterceptor();
- }
}
synchronized void initialize(final ManagedLedgerInitializeLedgerCallback callback, final Object ctx) {
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
index 0b96b5922e2..e390621c7c3 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
@@ -29,6 +29,7 @@ import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
import org.apache.bookkeeper.mledger.util.SafeRun;
import org.apache.bookkeeper.util.SafeRunnable;
@@ -63,6 +64,7 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
@SuppressWarnings("unused")
ByteBuf data;
private int dataLength;
+ private ManagedLedgerInterceptor.PayloadProcessorHandle payloadProcessorHandle = null;
private static final AtomicReferenceFieldUpdater<OpAddEntry, OpAddEntry.State> STATE_UPDATER = AtomicReferenceFieldUpdater
.newUpdater(OpAddEntry.class, OpAddEntry.State.class, "state");
@@ -105,6 +107,7 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
op.entryId = -1;
op.startTime = System.nanoTime();
op.state = State.OPEN;
+ op.payloadProcessorHandle = null;
ml.mbean.addAddEntrySample(op.dataLength);
return op;
}
@@ -125,6 +128,12 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
// internally asyncAddEntry() will take the ownership of the buffer and release it at the end
addOpCount = ManagedLedgerImpl.ADD_OP_COUNT_UPDATER.incrementAndGet(ml);
lastInitTime = System.nanoTime();
+ if (ml.getManagedLedgerInterceptor() != null) {
+ payloadProcessorHandle = ml.getManagedLedgerInterceptor().processPayloadBeforeLedgerWrite(this, duplicateBuffer);
+ if (payloadProcessorHandle != null) {
+ duplicateBuffer = payloadProcessorHandle.getProcessedPayload();
+ }
+ }
ledger.asyncAddEntry(duplicateBuffer, this, addOpCount);
} else {
log.warn("[{}] initiate with unexpected state {}, expect OPEN state.", ml.getName(), state);
@@ -137,6 +146,9 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
ReferenceCountUtil.release(data);
cb.addFailed(e, ctx);
ml.mbean.recordAddEntryError();
+ if (payloadProcessorHandle != null) {
+ payloadProcessorHandle.release();
+ }
}
}
@@ -177,6 +189,9 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
// Called in executor hashed on managed ledger name, once the add operation is complete
@Override
public void safeRun() {
+ if (payloadProcessorHandle != null) {
+ payloadProcessorHandle.release();
+ }
// Remove this entry from the head of the pending queue
OpAddEntry firstInQueue = ml.pendingAddEntries.poll();
if (firstInQueue == null) {
@@ -308,6 +323,10 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
return numberOfMessages;
}
+ public Object getCtx() {
+ return ctx;
+ }
+
public void setNumberOfMessages(int numberOfMessages) {
this.numberOfMessages = numberOfMessages;
}
@@ -343,6 +362,7 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
entryId = -1;
startTime = -1;
lastInitTime = -1;
+ payloadProcessorHandle = null;
recyclerHandle.recycle(this);
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/intercept/ManagedLedgerInterceptor.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/intercept/ManagedLedgerInterceptor.java
index 026d6ed6d18..13fc530529b 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/intercept/ManagedLedgerInterceptor.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/intercept/ManagedLedgerInterceptor.java
@@ -18,6 +18,7 @@
*/
package org.apache.bookkeeper.mledger.intercept;
+import io.netty.buffer.ByteBuf;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.common.annotation.InterfaceAudience;
import org.apache.bookkeeper.common.annotation.InterfaceStability;
@@ -58,4 +59,38 @@ public interface ManagedLedgerInterceptor {
* @param propertiesMap map of properties.
*/
void onUpdateManagedLedgerInfo(Map<String, String> propertiesMap);
+
+ /**
+ * A reference handle to the payload processor
+ */
+ interface PayloadProcessorHandle {
+ /**
+ * To obtain the processed data
+ * @return processed data
+ */
+ ByteBuf getProcessedPayload();
+
+ /**
+ * To release resources used in processor, if any
+ */
+ void release();
+ }
+ /**
+ * Intercept after entry is read from ledger, before it gets cached.
+ * @param dataReadFromLedger data from ledger
+ * @return handle to the processor
+ */
+ default PayloadProcessorHandle processPayloadBeforeEntryCache(ByteBuf dataReadFromLedger){
+ return null;
+ }
+
+ /**
+ * Intercept before payload gets written to ledger
+ * @param ledgerWriteOp OpAddEntry used to trigger ledger write.
+ * @param dataToBeStoredInLedger data to be stored in ledger
+ * @return handle to the processor
+ */
+ default PayloadProcessorHandle processPayloadBeforeLedgerWrite(OpAddEntry ledgerWriteOp, ByteBuf dataToBeStoredInLedger){
+ return null;
+ }
}
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index aed35bdad41..f3fe8e86441 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -24,6 +24,7 @@ import com.google.common.collect.Sets;
import io.netty.util.internal.PlatformDependent;
import java.util.ArrayList;
import java.util.HashSet;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Optional;
import java.util.Properties;
@@ -1056,6 +1057,11 @@ public class ServiceConfiguration implements PulsarConfiguration {
)
private boolean disableBrokerInterceptors = true;
+ @FieldContext(
+ category = CATEGORY_SERVER,
+ doc = "List of interceptors for payload processing.")
+ private Set<String> brokerEntryPayloadProcessors = new LinkedHashSet<>();
+
@FieldContext(
doc = "There are two policies to apply when broker metadata session expires: session expired happens, \"shutdown\" or \"reconnect\". \n\n"
+ " With \"shutdown\", the broker will be restarted.\n\n"
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 415d802d862..1e440b883da 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
@@ -18,16 +18,22 @@
*/
package org.apache.pulsar.broker.intercept;
+import io.netty.buffer.ByteBuf;
import java.io.IOException;
+import java.util.Map;
import javax.servlet.FilterChain;
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.Consumer;
+import org.apache.pulsar.broker.service.Producer;
import org.apache.pulsar.broker.service.ServerCnx;
import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.common.api.proto.BaseCommand;
+import org.apache.pulsar.common.api.proto.CommandAck;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.classification.InterfaceAudience;
import org.apache.pulsar.common.classification.InterfaceStability;
@@ -58,6 +64,65 @@ public interface BrokerInterceptor extends AutoCloseable {
MessageMetadata msgMetadata) {
}
+ /**
+ * Called by the broker when a new connection is created.
+ */
+ default void onConnectionCreated(ServerCnx cnx){
+ }
+
+ /**
+ * Called by the broker when a new connection is created.
+ */
+ default void producerCreated(ServerCnx cnx, Producer producer,
+ Map<String, String> metadata){
+ }
+
+ /**
+ * Intercept after a consumer is created.
+ *
+ * @param cnx client Connection
+ * @param consumer Consumer object
+ * @param metadata A map of metdata
+ */
+ default void consumerCreated(ServerCnx cnx,
+ Consumer consumer,
+ Map<String, String> metadata) {
+ }
+
+ /**
+ * Intercept after a message is produced.
+ *
+ * @param cnx client Connection
+ * @param producer Producer object
+ * @param publishContext Publish Context
+ */
+ default void messageProduced(ServerCnx cnx, Producer producer, long startTimeNs, long ledgerId,
+ long entryId, Topic.PublishContext publishContext) {
+ }
+
+ /**
+ * Intercept after a message is dispatched to consumer.
+ *
+ * @param cnx client Connection
+ * @param consumer Consumer object
+ * @param ledgerId Ledger ID
+ * @param entryId Entry ID
+ * @param headersAndPayload Data
+ */
+ default void messageDispatched(ServerCnx cnx, Consumer consumer, long ledgerId,
+ long entryId, ByteBuf headersAndPayload) {
+ }
+
+ /**
+ * Intercept after a message ack is processed.
+ *
+ * @param cnx client Connection
+ * @param ackCmd Command object
+ */
+ default void messageAcked(ServerCnx cnx, Consumer consumer,
+ CommandAck ackCmd) {
+ }
+
/**
* Called by the broker while new command incoming.
*/
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 6b06920a9bf..f990f7d49bc 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
@@ -18,7 +18,9 @@
*/
package org.apache.pulsar.broker.intercept;
+import io.netty.buffer.ByteBuf;
import java.io.IOException;
+import java.util.Map;
import javax.servlet.ServletException;
import javax.servlet.ServletRequest;
import javax.servlet.ServletResponse;
@@ -28,9 +30,13 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.broker.ClassLoaderSwitcher;
import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.Producer;
import org.apache.pulsar.broker.service.ServerCnx;
import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.common.api.proto.BaseCommand;
+import org.apache.pulsar.common.api.proto.CommandAck;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.intercept.InterceptException;
import org.apache.pulsar.common.nar.NarClassLoader;
@@ -57,6 +63,43 @@ public class BrokerInterceptorWithClassLoader implements BrokerInterceptor {
}
}
+ @Override
+ public void producerCreated(ServerCnx cnx, Producer producer,
+ Map<String, String> metadata){
+ this.interceptor.producerCreated(cnx, producer, metadata);
+ }
+
+ @Override
+ public void consumerCreated(ServerCnx cnx,
+ Consumer consumer,
+ Map<String, String> metadata) {
+ this.interceptor.consumerCreated(
+ cnx, consumer, metadata);
+ }
+
+ @Override
+ public void messageProduced(ServerCnx cnx, Producer producer, long startTimeNs, long ledgerId,
+ long entryId, Topic.PublishContext publishContext) {
+ this.interceptor.messageProduced(cnx, producer, startTimeNs, ledgerId, entryId, publishContext);
+ }
+
+ @Override
+ public void messageDispatched(ServerCnx cnx, Consumer consumer, long ledgerId,
+ long entryId, ByteBuf headersAndPayload) {
+ this.interceptor.messageDispatched(cnx, consumer, ledgerId, entryId, headersAndPayload);
+ }
+
+ @Override
+ public void messageAcked(ServerCnx cnx, Consumer consumer,
+ CommandAck ackCmd) {
+ this.interceptor.messageAcked(cnx, consumer, ackCmd);
+ }
+
+ @Override
+ public void onConnectionCreated(ServerCnx cnx) {
+ this.interceptor.onConnectionCreated(cnx);
+ }
+
@Override
public void onPulsarCommand(BaseCommand command, ServerCnx cnx) throws InterceptException {
try (ClassLoaderSwitcher ignored = new ClassLoaderSwitcher(classLoader)) {
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 22b5064bbd2..878f2cb5320 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
@@ -19,6 +19,7 @@
package org.apache.pulsar.broker.intercept;
import com.google.common.collect.ImmutableMap;
+import io.netty.buffer.ByteBuf;
import java.io.IOException;
import java.util.Map;
import javax.servlet.ServletException;
@@ -28,9 +29,13 @@ 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.Consumer;
+import org.apache.pulsar.broker.service.Producer;
import org.apache.pulsar.broker.service.ServerCnx;
import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.common.api.proto.BaseCommand;
+import org.apache.pulsar.common.api.proto.CommandAck;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.intercept.InterceptException;
@@ -102,6 +107,75 @@ public class BrokerInterceptors implements BrokerInterceptor {
}
}
+ @Override
+ public void consumerCreated(ServerCnx cnx,
+ Consumer consumer,
+ Map<String, String> metadata) {
+ if (interceptors == null || interceptors.isEmpty()) {
+ return;
+ }
+ for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ value.consumerCreated(
+ cnx,
+ consumer,
+ metadata);
+ }
+ }
+
+ @Override
+ public void producerCreated(ServerCnx cnx, Producer producer,
+ Map<String, String> metadata){
+ if (interceptors == null || interceptors.isEmpty()) {
+ return;
+ }
+ for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ value.producerCreated(cnx, producer, metadata);
+ }
+ }
+
+ @Override
+ public void messageProduced(ServerCnx cnx, Producer producer, long startTimeNs, long ledgerId,
+ long entryId, Topic.PublishContext publishContext) {
+ if (interceptors == null || interceptors.isEmpty()) {
+ return;
+ }
+ for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ value.messageProduced(cnx, producer, startTimeNs, ledgerId, entryId, publishContext);
+ }
+ }
+
+ @Override
+ public void messageDispatched(ServerCnx cnx, Consumer consumer, long ledgerId,
+ long entryId, ByteBuf headersAndPayload) {
+ if (interceptors == null || interceptors.isEmpty()) {
+ return;
+ }
+ for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ value.messageDispatched(cnx, consumer, ledgerId, entryId, headersAndPayload);
+ }
+ }
+
+ @Override
+ public void messageAcked(ServerCnx cnx, Consumer consumer,
+ CommandAck ackCmd) {
+ if (interceptors == null || interceptors.isEmpty()) {
+ return;
+ }
+ for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ value.messageAcked(cnx, consumer, ackCmd);
+ }
+ }
+
+ @Override
+ public void onConnectionCreated(ServerCnx cnx) {
+ if (interceptors == null || interceptors.isEmpty()) {
+ return;
+ }
+ for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ value.onConnectionCreated(cnx);
+ }
+ }
+
@Override
public void onPulsarCommand(BaseCommand command, ServerCnx cnx) throws InterceptException {
for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImpl.java
index 424797fa52a..0f634e0553e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImpl.java
@@ -18,6 +18,8 @@
*/
package org.apache.pulsar.broker.intercept;
+import io.netty.buffer.ByteBuf;
+import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -25,9 +27,11 @@ import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.client.api.LedgerEntry;
import org.apache.bookkeeper.mledger.impl.OpAddEntry;
import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
+import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.pulsar.common.api.proto.BrokerEntryMetadata;
import org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor;
import org.apache.pulsar.common.intercept.BrokerEntryMetadataInterceptor;
+import org.apache.pulsar.common.intercept.ManagedLedgerPayloadProcessor;
import org.apache.pulsar.common.protocol.Commands;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,9 +40,23 @@ public class ManagedLedgerInterceptorImpl implements ManagedLedgerInterceptor {
private static final Logger log = LoggerFactory.getLogger(ManagedLedgerInterceptorImpl.class);
private static final String INDEX = "index";
private final Set<BrokerEntryMetadataInterceptor> brokerEntryMetadataInterceptors;
+ private final Set<ManagedLedgerPayloadProcessor.Processor> inputProcessors;
+ private final Set<ManagedLedgerPayloadProcessor.Processor> outputProcessors;
- public ManagedLedgerInterceptorImpl(Set<BrokerEntryMetadataInterceptor> brokerEntryMetadataInterceptors) {
+ public ManagedLedgerInterceptorImpl(Set<BrokerEntryMetadataInterceptor> brokerEntryMetadataInterceptors,
+ Set<ManagedLedgerPayloadProcessor> brokerEntryPayloadProcessors) {
this.brokerEntryMetadataInterceptors = brokerEntryMetadataInterceptors;
+ if (brokerEntryPayloadProcessors != null) {
+ this.inputProcessors = new LinkedHashSet<>();
+ this.outputProcessors = new LinkedHashSet<>();
+ for (ManagedLedgerPayloadProcessor processor : brokerEntryPayloadProcessors) {
+ this.inputProcessors.add(processor.inputProcessor());
+ this.outputProcessors.add(processor.outputProcessor());
+ }
+ } else {
+ this.inputProcessors = null;
+ this.outputProcessors = null;
+ }
}
public long getIndex() {
@@ -129,4 +147,47 @@ public class ManagedLedgerInterceptorImpl implements ManagedLedgerInterceptor {
}
}
}
+
+ private PayloadProcessorHandle processPayload(Set<ManagedLedgerPayloadProcessor.Processor> processors,
+ Object context, ByteBuf payload) {
+
+ ByteBuf tmpData = payload;
+ final Set<ImmutablePair<ManagedLedgerPayloadProcessor.Processor, ByteBuf>> processedSet = new LinkedHashSet<>();
+ for (ManagedLedgerPayloadProcessor.Processor payloadProcessor : processors) {
+ if (payloadProcessor != null) {
+ tmpData = payloadProcessor.process(context, tmpData);
+ processedSet.add(new ImmutablePair<>(payloadProcessor, tmpData));
+ }
+ }
+ final ByteBuf dataToReturn = tmpData;
+ return new PayloadProcessorHandle() {
+ @Override
+ public ByteBuf getProcessedPayload() {
+ return dataToReturn;
+ }
+
+ @Override
+ public void release() {
+ for (ImmutablePair<ManagedLedgerPayloadProcessor.Processor, ByteBuf> p : processedSet) {
+ p.left.release(p.right);
+ }
+ processedSet.clear();
+ }
+ };
+ }
+ @Override
+ public PayloadProcessorHandle processPayloadBeforeLedgerWrite(OpAddEntry op, ByteBuf ledgerData) {
+ if (this.inputProcessors == null || this.inputProcessors.size() == 0) {
+ return null;
+ }
+ return processPayload(this.inputProcessors, op.getCtx(), ledgerData);
+ }
+
+ @Override
+ public PayloadProcessorHandle processPayloadBeforeEntryCache(ByteBuf ledgerData){
+ if (this.outputProcessors == null || this.outputProcessors.size() == 0) {
+ return null;
+ }
+ return processPayload(this.outputProcessors, null, ledgerData);
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index e23cb663dec..dfd8db1a732 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -134,6 +134,7 @@ import org.apache.pulsar.common.events.EventsTopicNames;
import org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor;
import org.apache.pulsar.common.intercept.BrokerEntryMetadataInterceptor;
import org.apache.pulsar.common.intercept.BrokerEntryMetadataUtils;
+import org.apache.pulsar.common.intercept.ManagedLedgerPayloadProcessor;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicDomain;
@@ -271,6 +272,7 @@ public class BrokerService implements Closeable {
private BrokerInterceptor interceptor;
private Set<BrokerEntryMetadataInterceptor> brokerEntryMetadataInterceptors;
+ private Set<ManagedLedgerPayloadProcessor> brokerEntryPayloadProcessors;
public BrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws Exception {
this.pulsar = pulsar;
@@ -377,6 +379,9 @@ public class BrokerService implements Closeable {
.loadBrokerEntryMetadataInterceptors(pulsar.getConfiguration().getBrokerEntryMetadataInterceptors(),
BrokerService.class.getClassLoader());
+ this.brokerEntryPayloadProcessors = BrokerEntryMetadataUtils.loadInterceptors(pulsar.getConfiguration()
+ .getBrokerEntryPayloadProcessors(), BrokerService.class.getClassLoader());
+
this.bundlesQuotas = new BundlesQuotas(pulsar.getLocalMetadataStore());
}
@@ -1338,7 +1343,7 @@ public class BrokerService implements Closeable {
: CompletableFuture.completedFuture(null);
maxTopicsCheck.thenCompose(__ -> getManagedLedgerConfig(topicName)).thenAccept(managedLedgerConfig -> {
- if (isBrokerEntryMetadataEnabled()) {
+ if (isBrokerEntryMetadataEnabled() || isBrokerPayloadProcessorEnabled()) {
// init managedLedger interceptor
Set<BrokerEntryMetadataInterceptor> interceptors = new HashSet<>();
for (BrokerEntryMetadataInterceptor interceptor : brokerEntryMetadataInterceptors) {
@@ -1349,9 +1354,9 @@ public class BrokerService implements Closeable {
interceptors.add(interceptor);
}
}
- managedLedgerConfig.setManagedLedgerInterceptor(new ManagedLedgerInterceptorImpl(interceptors));
+ managedLedgerConfig.setManagedLedgerInterceptor(
+ new ManagedLedgerInterceptorImpl(interceptors, brokerEntryPayloadProcessors));
}
-
managedLedgerConfig.setCreateIfMissing(createIfMissing);
// Once we have the configuration, we can proceed with the async open operation
@@ -2878,6 +2883,10 @@ public class BrokerService implements Closeable {
return !brokerEntryMetadataInterceptors.isEmpty();
}
+ public boolean isBrokerPayloadProcessorEnabled() {
+ return !brokerEntryPayloadProcessors.isEmpty();
+ }
+
public void pausedConnections(int numberOfConnections) {
pausedConnections.add(numberOfConnections);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
index 1659d25c74e..ba6b3c3c297 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
@@ -1017,6 +1017,10 @@ public class Consumer {
return startMessageId;
}
+ public Map<String, String> getMetadata() {
+ return metadata;
+ }
+
private int getStickyKeyHash(Entry entry) {
byte[] stickyKey = Commands.peekStickyKey(entry.getDataBuffer(), topicName, subscription.getName());
return StickyKeyConsumerSelector.makeStickyKeyHash(stickyKey);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
index 8043695bb99..8631bd2bbe6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
@@ -29,6 +29,7 @@ import io.netty.buffer.ByteBuf;
import io.netty.util.Recycler;
import io.netty.util.Recycler.Handle;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
@@ -317,6 +318,11 @@ public class Producer {
}
private static final class MessagePublishContext implements PublishContext, Runnable {
+ /*
+ * To store context information built by message payload
+ * processors (time duration, size etc), if any configured
+ */
+ Map<String, Object> propertyMap;
private Producer producer;
private long sequenceId;
private long ledgerId;
@@ -347,6 +353,23 @@ public class Producer {
return chunked;
}
+ @Override
+ public void setProperty(String propertyName, Object value){
+ if (this.propertyMap == null) {
+ this.propertyMap = new HashMap<>();
+ }
+ this.propertyMap.put(propertyName, value);
+ }
+
+ @Override
+ public Object getProperty(String propertyName){
+ if (this.propertyMap != null) {
+ return this.propertyMap.get(propertyName);
+ } else {
+ return null;
+ }
+ }
+
@Override
public long getHighestSequenceId() {
return highestSequenceId;
@@ -446,6 +469,10 @@ public class Producer {
producer.chunkedMessageRate.recordEvent();
}
producer.publishOperationCompleted();
+ if (producer.cnx.getBrokerService().getInterceptor() != null){
+ producer.cnx.getBrokerService().getInterceptor().messageProduced(
+ (ServerCnx) producer.cnx, producer, startTimeNs, ledgerId, entryId, this);
+ }
recycle();
}
@@ -462,6 +489,9 @@ public class Producer {
callback.originalSequenceId = -1L;
callback.startTimeNs = startTimeNs;
callback.isMarker = isMarker;
+ if (callback.propertyMap != null) {
+ callback.propertyMap.clear();
+ }
return callback;
}
@@ -479,6 +509,9 @@ public class Producer {
callback.startTimeNs = startTimeNs;
callback.chunked = chunked;
callback.isMarker = isMarker;
+ if (callback.propertyMap != null) {
+ callback.propertyMap.clear();
+ }
return callback;
}
@@ -518,6 +551,9 @@ public class Producer {
startTimeNs = -1L;
chunked = false;
isMarker = false;
+ if (propertyMap != null) {
+ propertyMap.clear();
+ }
recyclerHandle.recycle(this);
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index 24cedf27277..7d10a43cfe2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -634,6 +634,10 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
if (isNotBlank(clientVersion) && !clientVersion.contains(" ") /* ignore default version: pulsar client */) {
this.clientVersion = clientVersion.intern();
}
+ BrokerInterceptor brokerInterceptor = getBrokerService().getInterceptor();
+ if (brokerInterceptor != null) {
+ brokerInterceptor.onConnectionCreated(this);
+ }
}
// According to auth result, send newConnected or newAuthChallenge command.
@@ -1062,6 +1066,9 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
log.info("[{}] Created subscription on topic {} / {}",
remoteAddress, topicName, subscriptionName);
commandSender.sendSuccessResponse(requestId);
+ if (getBrokerService().getInterceptor() != null){
+ getBrokerService().getInterceptor().consumerCreated(this, consumer, metadata);
+ }
} else {
// The consumer future was completed before by a close command
try {
@@ -1336,6 +1343,10 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
commandSender.sendProducerSuccessResponse(requestId, producerName,
producer.getLastSequenceId(), producer.getSchemaVersion(),
newTopicEpoch, true /* producer is ready now */);
+ if (getBrokerService().getInterceptor() != null) {
+ getBrokerService().getInterceptor().
+ producerCreated(this, producer, metadata);
+ }
return;
} else {
// The producer's future was completed before by
@@ -1382,6 +1393,10 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
commandSender.sendProducerSuccessResponse(requestId, producerName,
producer.getLastSequenceId(), producer.getSchemaVersion(),
Optional.empty(), false/* producer is not ready now */);
+ if (getBrokerService().getInterceptor() != null) {
+ getBrokerService().getInterceptor().
+ producerCreated(this, producer, metadata);
+ }
}
});
}
@@ -1462,11 +1477,15 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
final long consumerId = ack.getConsumerId();
if (consumerFuture != null && consumerFuture.isDone() && !consumerFuture.isCompletedExceptionally()) {
- consumerFuture.getNow(null).messageAcked(ack).thenRun(() -> {
+ Consumer consumer = consumerFuture.getNow(null);
+ consumer.messageAcked(ack).thenRun(() -> {
if (hasRequestId) {
ctx.writeAndFlush(Commands.newAckResponse(
requestId, null, null, consumerId));
}
+ if (getBrokerService().getInterceptor() != null) {
+ getBrokerService().getInterceptor().messageAcked(this, consumer, ack);
+ }
}).exceptionally(e -> {
if (hasRequestId) {
ctx.writeAndFlush(Commands.newAckResponse(requestId,
@@ -2687,6 +2706,12 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
val brokerInterceptor = getBrokerService().getInterceptor();
if (brokerInterceptor != null) {
brokerInterceptor.onPulsarCommand(command, this);
+
+ CompletableFuture<Consumer> consumerFuture = consumers.get(consumerId);
+ if (consumerFuture != null && consumerFuture.isDone() && !consumerFuture.isCompletedExceptionally()) {
+ Consumer consumer = consumerFuture.getNow(null);
+ brokerInterceptor.messageDispatched(this, consumer, ledgerId, entryId, metadataAndPayload);
+ }
} else {
log.debug("BrokerInterceptor is not set in newMessageAndIntercept");
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
index 14e8e3fe004..131fd2e933c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
@@ -98,6 +98,13 @@ public interface Topic {
default boolean isMarkerMessage() {
return false;
}
+
+ default void setProperty(String propertyName, Object value) {
+ }
+
+ default Object getProperty(String propertyName) {
+ return null;
+ }
}
CompletableFuture<Void> initialize();
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 977415525d8..d0c163f4356 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
@@ -123,6 +123,34 @@ public class BrokerInterceptorTest extends ProducerConsumerBase {
Assert.assertTrue(((CounterBrokerInterceptor)listener).getCount() >= 2);
}
+ @Test
+ public void testConnectionCreation() throws PulsarClientException {
+ BrokerInterceptor listener = pulsar.getBrokerInterceptor();
+ Assert.assertTrue(listener instanceof CounterBrokerInterceptor);
+ pulsarClient.newProducer(Schema.BOOL).topic("test").create();
+ pulsarClient.newConsumer(Schema.STRING).topic("test1").subscriptionName("test-sub").subscribe();
+ // single connection for both producer and consumer
+ Assert.assertTrue(((CounterBrokerInterceptor)listener).getConnectionCreationCount() == 1);
+ }
+
+ @Test
+ public void testProducerCreation() throws PulsarClientException {
+ BrokerInterceptor listener = pulsar.getBrokerInterceptor();
+ Assert.assertTrue(listener instanceof CounterBrokerInterceptor);
+ Assert.assertTrue(((CounterBrokerInterceptor)listener).getProducerCount() == 0);
+ pulsarClient.newProducer(Schema.BOOL).topic("test").create();
+ Assert.assertTrue(((CounterBrokerInterceptor)listener).getProducerCount() == 1);
+ }
+
+ @Test
+ public void testConsumerCreation() throws PulsarClientException {
+ BrokerInterceptor listener = pulsar.getBrokerInterceptor();
+ Assert.assertTrue(listener instanceof CounterBrokerInterceptor);
+ Assert.assertTrue(((CounterBrokerInterceptor)listener).getConsumerCount() == 0);
+ pulsarClient.newConsumer(Schema.STRING).topic("test1").subscriptionName("test-sub").subscribe();
+ Assert.assertTrue(((CounterBrokerInterceptor)listener).getConsumerCount() == 1);
+ }
+
@Test
public void testBeforeSendMessage() throws PulsarClientException {
BrokerInterceptor listener = pulsar.getBrokerInterceptor();
@@ -138,13 +166,17 @@ public class BrokerInterceptorTest extends ProducerConsumerBase {
.subscriptionName("test")
.subscribe();
+ assertEquals(((CounterBrokerInterceptor)listener).getMessagePublishCount(),0);
+ assertEquals(((CounterBrokerInterceptor)listener).getMessageDispatchCount(),0);
producer.send("hello world");
+ assertEquals(((CounterBrokerInterceptor)listener).getMessagePublishCount(),1);
Message<String> msg = consumer.receive();
assertEquals(msg.getValue(), "hello world");
assertEquals(((CounterBrokerInterceptor) listener).getBeforeSendCount(), 1);
+ assertEquals(((CounterBrokerInterceptor)listener).getMessageDispatchCount(),1);
}
@Test
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 1462cfab895..3b0044386cc 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
@@ -18,9 +18,11 @@
*/
package org.apache.pulsar.broker.intercept;
+import io.netty.buffer.ByteBuf;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import javax.servlet.FilterChain;
import javax.servlet.ServletException;
import javax.servlet.ServletRequest;
@@ -31,17 +33,30 @@ import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.Producer;
import org.apache.pulsar.broker.service.ServerCnx;
import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.common.api.proto.BaseCommand;
import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.api.proto.CommandAck;
+import org.apache.pulsar.common.stats.Rate;
import org.eclipse.jetty.server.Response;
+
@Slf4j
public class CounterBrokerInterceptor implements BrokerInterceptor {
int beforeSendCount = 0;
int count = 0;
+ int connectionCreationCount = 0;
+ int producerCount = 0;
+ int consumerCount = 0;
+ int messageCount = 0;
+ int messageDispatchCount = 0;
+ int messageAckCount = 0;
+
private List<ResponseEvent> responseList = new ArrayList<>();
@Data
@@ -51,6 +66,52 @@ public class CounterBrokerInterceptor implements BrokerInterceptor {
private int responseStatus;
}
+ @Override
+ public void onConnectionCreated(ServerCnx cnx){
+ log.info("Connection created {}", cnx);
+ connectionCreationCount++;
+ }
+
+ @Override
+ public void producerCreated(ServerCnx cnx, Producer producer,
+ Map<String, String> metadata){
+ log.info("Producer created with name={}, id={}",
+ producer.getProducerName(), producer.getProducerId());
+ producerCount++;
+ }
+
+ @Override
+ public void consumerCreated(ServerCnx cnx,
+ Consumer consumer,
+ Map<String, String> metadata) {
+ log.info("Consumer created with name={}, id={}",
+ consumer.consumerName(), consumer.consumerId());
+ consumerCount++;
+ }
+
+ @Override
+ public void messageProduced(ServerCnx cnx, Producer producer, long startTimeNs, long ledgerId,
+ long entryId,
+ Topic.PublishContext publishContext) {
+ log.info("Message published topic={}, producer={}",
+ producer.getTopic().getName(), producer.getProducerName());
+ messageCount++;
+ }
+
+ @Override
+ public void messageDispatched(ServerCnx cnx, Consumer consumer, long ledgerId,
+ long entryId, ByteBuf headersAndPayload) {
+ log.info("Message dispatched topic={}, consumer={}",
+ consumer.getSubscription().getTopic().getName(), consumer.consumerName());
+ messageDispatchCount++;
+ }
+
+ @Override
+ public void messageAcked(ServerCnx cnx, Consumer consumer,
+ CommandAck ack) {
+ messageAckCount++;
+ }
+
@Override
public void beforeSendMessage(Subscription subscription,
Entry entry,
@@ -109,10 +170,34 @@ public class CounterBrokerInterceptor implements BrokerInterceptor {
return count;
}
+ public int getProducerCount() {
+ return producerCount;
+ }
+
+ public int getConsumerCount() {
+ return consumerCount;
+ }
+
+ public int getMessagePublishCount() {
+ return messageCount;
+ }
+
+ public int getMessageDispatchCount() {
+ return messageDispatchCount;
+ }
+
+ public int getMessageAckCount() {
+ return messageAckCount;
+ }
+
public int getBeforeSendCount() {
return beforeSendCount;
}
+ public int getConnectionCreationCount() {
+ return connectionCreationCount;
+ }
+
public void clearResponseList() {
responseList.clear();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/MangedLedgerInterceptorImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/MangedLedgerInterceptorImplTest.java
index 75dc8917193..60067a7016c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/MangedLedgerInterceptorImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/MangedLedgerInterceptorImplTest.java
@@ -18,6 +18,8 @@
*/
package org.apache.pulsar.broker.intercept;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
import lombok.Cleanup;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -31,10 +33,12 @@ import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.pulsar.common.api.proto.BrokerEntryMetadata;
import org.apache.pulsar.common.intercept.BrokerEntryMetadataInterceptor;
import org.apache.pulsar.common.intercept.BrokerEntryMetadataUtils;
+import org.apache.pulsar.common.intercept.ManagedLedgerPayloadProcessor;
import org.apache.pulsar.common.protocol.Commands;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.testng.Assert;
import org.testng.annotations.Test;
import java.nio.charset.StandardCharsets;
@@ -50,13 +54,54 @@ import static org.testng.Assert.assertNotNull;
public class MangedLedgerInterceptorImplTest extends MockedBookKeeperTestCase {
private static final Logger log = LoggerFactory.getLogger(MangedLedgerInterceptorImplTest.class);
+ public class TestPayloadProcessor implements ManagedLedgerPayloadProcessor {
+ @Override
+ public Processor inputProcessor() {
+ return new Processor() {
+ @Override
+ public ByteBuf process(Object contextObj, ByteBuf inputPayload) {
+ byte[] newMessage = (new String("Modified Test Message")).getBytes();
+ ByteBuf processedPayload = Unpooled.wrappedBuffer(newMessage, 0, newMessage.length);
+ inputPayload.release();
+ return processedPayload.retainedDuplicate();
+ }
+
+ @Override
+ public void release(ByteBuf processedPayload) {
+ processedPayload.release();
+ }
+ };
+ }
+ @Override
+ public Processor outputProcessor() {
+ return new Processor() {
+ @Override
+ public ByteBuf process(Object contextObj, ByteBuf inputPayload) {
+ byte[] bytes = new byte[inputPayload.readableBytes()];
+ inputPayload.readBytes(bytes);
+ String storedMessage = new String(bytes);
+ Assert.assertTrue(storedMessage.equals("Modified Test Message"));
+
+ byte[] newMessage = (new String("Test Message")).getBytes();
+ inputPayload.release();
+ return Unpooled.wrappedBuffer(newMessage, 0, newMessage.length).retainedDuplicate();
+ }
+
+ @Override
+ public void release(ByteBuf processedPayload) {
+ processedPayload.release();
+ }
+ };
+ }
+ }
+
@Test
public void testAddBrokerEntryMetadata() throws Exception {
final int MOCK_BATCH_SIZE = 2;
int numberOfEntries = 10;
final String ledgerAndCursorName = "topicEntryMetadataSequenceId";
- ManagedLedgerInterceptor interceptor = new ManagedLedgerInterceptorImpl(getBrokerEntryMetadataInterceptors());
+ ManagedLedgerInterceptor interceptor = new ManagedLedgerInterceptorImpl(getBrokerEntryMetadataInterceptors(),null);
ManagedLedgerConfig config = new ManagedLedgerConfig();
config.setMaxEntriesPerLedger(2);
@@ -82,11 +127,35 @@ public class MangedLedgerInterceptorImplTest extends MockedBookKeeperTestCase {
ledger.close();
factory.shutdown();
}
+ @Test
+ public void testMessagePayloadProcessor() throws Exception {
+ final String ledgerAndCursorName = "topicEntryWithPayloadProcessed";
+
+ Set<ManagedLedgerPayloadProcessor> processors = new HashSet();
+ processors.add(new TestPayloadProcessor());
+ ManagedLedgerInterceptor interceptor = new ManagedLedgerInterceptorImpl(new HashSet(),processors);
+
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMaxEntriesPerLedger(2);
+ config.setManagedLedgerInterceptor(interceptor);
+
+ ManagedLedger ledger = factory.open(ledgerAndCursorName, config);
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor(ledgerAndCursorName);
+ ledger.addEntry("Test Message".getBytes());
+ factory.getEntryCacheManager().clear();
+ List<Entry> entryList = cursor.readEntries(1);
+ String message = new String(entryList.get(0).getData());
+ Assert.assertTrue(message.equals("Test Message"));
+ cursor.close();
+ ledger.close();
+ factory.shutdown();
+ config.setManagedLedgerInterceptor(null);
+ }
@Test(timeOut = 20000)
public void testRecoveryIndex() throws Exception {
final int MOCK_BATCH_SIZE = 2;
- ManagedLedgerInterceptor interceptor = new ManagedLedgerInterceptorImpl(getBrokerEntryMetadataInterceptors());
+ ManagedLedgerInterceptor interceptor = new ManagedLedgerInterceptorImpl(getBrokerEntryMetadataInterceptors(),null);
ManagedLedgerConfig config = new ManagedLedgerConfig();
config.setManagedLedgerInterceptor(interceptor);
@@ -128,7 +197,7 @@ public class MangedLedgerInterceptorImplTest extends MockedBookKeeperTestCase {
final int MOCK_BATCH_SIZE = 2;
final int maxEntriesPerLedger = 5;
int maxSequenceIdPerLedger = MOCK_BATCH_SIZE * maxEntriesPerLedger;
- ManagedLedgerInterceptor interceptor = new ManagedLedgerInterceptorImpl(getBrokerEntryMetadataInterceptors());
+ ManagedLedgerInterceptor interceptor = new ManagedLedgerInterceptorImpl(getBrokerEntryMetadataInterceptors(),null);
ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/intercept/BrokerEntryMetadataUtils.java b/pulsar-common/src/main/java/org/apache/pulsar/common/intercept/BrokerEntryMetadataUtils.java
index 88f5972c027..09286e67900 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/intercept/BrokerEntryMetadataUtils.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/intercept/BrokerEntryMetadataUtils.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.common.intercept;
import java.lang.reflect.InvocationTargetException;
import java.util.HashSet;
import java.util.Set;
+import java.util.LinkedHashSet;
import org.apache.pulsar.common.util.ClassLoaderUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,7 +30,7 @@ import org.slf4j.LoggerFactory;
/**
* A tool class for loading BrokerEntryMetadataInterceptor classes.
*/
-public class BrokerEntryMetadataUtils {
+public class BrokerEntryMetadataUtils<T> {
private static final Logger log = LoggerFactory.getLogger(BrokerEntryMetadataUtils.class);
@@ -57,4 +58,28 @@ public class BrokerEntryMetadataUtils {
}
return interceptors;
}
+ public static <T> Set<T> loadInterceptors(
+ Set<String> interceptorNames, ClassLoader classLoader) {
+ Set<T> interceptors = new LinkedHashSet<>();
+ if (interceptorNames != null && interceptorNames.size() > 0) {
+ for (String interceptorName : interceptorNames) {
+ try {
+ Class<T> clz = (Class<T>) ClassLoaderUtils
+ .loadClass(interceptorName, classLoader);
+ try {
+ interceptors.add(clz.getDeclaredConstructor().newInstance());
+ } catch (InstantiationException | IllegalAccessException
+ | InvocationTargetException | NoSuchMethodException e) {
+ log.error("Create new instance for {} failed. Exception is {}",
+ interceptorName, e);
+ throw new RuntimeException(e);
+ }
+ } catch (ClassNotFoundException e) {
+ log.error("Load class for {} failed. Exception is {}", interceptorName, e);
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ return interceptors;
+ }
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/intercept/ManagedLedgerPayloadProcessor.java b/pulsar-common/src/main/java/org/apache/pulsar/common/intercept/ManagedLedgerPayloadProcessor.java
new file mode 100644
index 00000000000..94dfb73ea2c
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/intercept/ManagedLedgerPayloadProcessor.java
@@ -0,0 +1,62 @@
+/**
+ * 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.common.intercept;
+
+import io.netty.buffer.ByteBuf;
+
+public interface ManagedLedgerPayloadProcessor {
+ interface Processor {
+ /**
+ * Process the input payload and return a new payload
+ * NOTE: If this processor returns a different ByteBuf instance than the passed one
+ * DO THE FOLLOWING to avoid memory leaks
+ * 1. Call inputPayload.release() to release a reference
+ * 2. Call retain() on the ByteBuf that is being returned
+ * @param contextObj context object
+ * @param inputPayload The input payload buffer
+ * @return processed data
+ */
+ ByteBuf process(Object contextObj, ByteBuf inputPayload);
+
+ /**
+ * To release any resource used during the process.
+ * NOTE: To avoid memory leak, do the following ONLY if a different ByteBuf instance was returned in process()
+ * 1. Call processedPayload.release() to release a reference
+ * 2. Do any other cleanup needed
+ *
+ * @param processedPayload The processed payload that was returned in process() call
+ */
+ void release(ByteBuf processedPayload);
+ }
+ /**
+ * Used by ManagedLedger for pre-processing payload before storing in bookkeeper ledger
+ * @return Handle to Processor instance
+ */
+ default Processor inputProcessor() {
+ return null;
+ }
+
+ /**
+ * Used by ManagedLedger for processing payload after reading from bookkeeper ledger
+ * @return Handle to Processor instance
+ */
+ default Processor outputProcessor() {
+ return null;
+ }
+}