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:53 UTC
[pulsar] branch branch-2.9 updated (ef3fa9e4f11 -> b2e9c2cd286)
This is an automated email from the ASF dual-hosted git repository.
bogong pushed a change to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from ef3fa9e4f11 [fix][flaky-test]AdminApi2Test.testDeleteNamespace (#17157)
new c7124bb1325 [pulsar-broker] Broker extensions to allow operators of enterprise wide cluster better control and flexibility (#12536)
new b2e9c2cd286 [fix][flay-test]BrokerInterceptorTest.testProducerCreation (#17159)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../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 | 40 +++++++++-
.../broker/intercept/CounterBrokerInterceptor.java | 85 ++++++++++++++++++++++
.../intercept/MangedLedgerInterceptorImplTest.java | 75 ++++++++++++++++++-
.../common/intercept/BrokerEntryMetadataUtils.java | 27 ++++++-
.../intercept/ManagedLedgerPayloadProcessor.java | 62 ++++++++++++++++
20 files changed, 707 insertions(+), 19 deletions(-)
create mode 100644 pulsar-common/src/main/java/org/apache/pulsar/common/intercept/ManagedLedgerPayloadProcessor.java
[pulsar] 02/02: [fix][flay-test]BrokerInterceptorTest.testProducerCreation (#17159)
Posted by bo...@apache.org.
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 b2e9c2cd286124db86d54c36c4db014ada7d4c38
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Fri Aug 19 07:43:47 2022 +0800
[fix][flay-test]BrokerInterceptorTest.testProducerCreation (#17159)
(cherry picked from commit 84968e84c109ccc690fdd5779d2f09a50cd9da24)
---
.../pulsar/broker/intercept/BrokerInterceptorTest.java | 16 +++++++++-------
1 file changed, 9 insertions(+), 7 deletions(-)
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 d0c163f4356..0e46b147e7b 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
@@ -32,6 +32,7 @@ import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.nar.NarClassLoader;
+import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -111,7 +112,7 @@ public class BrokerInterceptorTest extends ProducerConsumerBase {
BrokerInterceptor listener = pulsar.getBrokerInterceptor();
Assert.assertTrue(listener instanceof CounterBrokerInterceptor);
admin.namespaces().createNamespace("public/test", 4);
- Assert.assertTrue(((CounterBrokerInterceptor)listener).getCount() >= 1);
+ Awaitility.await().until(() -> ((CounterBrokerInterceptor) listener).getCount() >= 1);
}
@Test
@@ -120,7 +121,7 @@ public class BrokerInterceptorTest extends ProducerConsumerBase {
Assert.assertTrue(listener instanceof CounterBrokerInterceptor);
pulsarClient.newProducer(Schema.BOOL).topic("test").create();
// CONNECT and PRODUCER
- Assert.assertTrue(((CounterBrokerInterceptor)listener).getCount() >= 2);
+ Awaitility.await().until(() -> ((CounterBrokerInterceptor) listener).getCount() >= 2);
}
@Test
@@ -130,7 +131,7 @@ public class BrokerInterceptorTest extends ProducerConsumerBase {
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);
+ Awaitility.await().until(() -> ((CounterBrokerInterceptor) listener).getConnectionCreationCount() == 1);
}
@Test
@@ -139,7 +140,7 @@ public class BrokerInterceptorTest extends ProducerConsumerBase {
Assert.assertTrue(listener instanceof CounterBrokerInterceptor);
Assert.assertTrue(((CounterBrokerInterceptor)listener).getProducerCount() == 0);
pulsarClient.newProducer(Schema.BOOL).topic("test").create();
- Assert.assertTrue(((CounterBrokerInterceptor)listener).getProducerCount() == 1);
+ Awaitility.await().until(() -> ((CounterBrokerInterceptor) listener).getProducerCount() == 1);
}
@Test
@@ -148,7 +149,7 @@ public class BrokerInterceptorTest extends ProducerConsumerBase {
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);
+ Awaitility.await().until(() -> ((CounterBrokerInterceptor) listener).getConsumerCount() == 1);
}
@Test
@@ -175,8 +176,8 @@ public class BrokerInterceptorTest extends ProducerConsumerBase {
assertEquals(msg.getValue(), "hello world");
- assertEquals(((CounterBrokerInterceptor) listener).getBeforeSendCount(), 1);
- assertEquals(((CounterBrokerInterceptor)listener).getMessageDispatchCount(),1);
+ Awaitility.await().until(() -> ((CounterBrokerInterceptor) listener).getBeforeSendCount() == 1);
+ Awaitility.await().until(() -> ((CounterBrokerInterceptor) listener).getMessageDispatchCount() == 1);
}
@Test
@@ -205,6 +206,7 @@ public class BrokerInterceptorTest extends ProducerConsumerBase {
}
});
future.get();
+ Awaitility.await().until(() -> !interceptor.getResponseList().isEmpty());
CounterBrokerInterceptor.ResponseEvent responseEvent = interceptor.getResponseList().get(0);
Assert.assertEquals(responseEvent.getRequestUri(), "/admin/v3/test/asyncGet/my-topic/1000");
Assert.assertEquals(responseEvent.getResponseStatus(),
[pulsar] 01/02: [pulsar-broker] Broker extensions to allow operators of enterprise wide cluster better control and flexibility (#12536)
Posted by bo...@apache.org.
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;
+ }
+}