[pulsar-broker] Broker extensions to allow operators of enterprise wide cluster better control and flexibility #12536

madhavan-narayanan · 2021-10-29T12:49:15Z

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 *

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): no
The public API: no
The schema: no
The default values of configurations: no
The wire protocol: no
The rest endpoints: no
The admin cli options: no
Anything that affects deployment: no

Documentation

Check the box below and label this PR (if you have committer privilege).

Need to update docs?

doc-required

(If you need help on updating docs, create a doc issue)
no-need-doc

(Please explain why)
doc

(If this PR contains doc changes)

…trol when operating org-wide messaging platform

github-actions · 2021-10-29T12:49:37Z

@madhavan-narayanan:Thanks for providing doc info!

github-actions · 2021-10-29T12:52:55Z

@madhavan-narayanan:Thanks for providing doc info!

github-actions · 2021-10-29T12:52:56Z

@madhavan-narayanan:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

github-actions · 2021-10-29T18:52:35Z

@madhavan-narayanan:Thanks for providing doc info!

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

Jason918 · 2021-10-30T02:25:35Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java

+        return create(ledgerEntry,null);
+    }
+
+    public static EntryImpl create(LedgerEntry ledgerEntry, ManagedLedgerInterceptor managedLedgerInterceptor) {


Put the 'managedLedgerInterceptor' in here seem a little strange.
IMHO, it's better put outside of create, the managedLedgerInterceptor processes the ledgerEntry and then pass the result to this create method.

I extended the create() to ensure that future usages of this method from other places don't miss out the interceptor.
I do agree that it will be clean to have the calling methods process the ledger entry using interceptors and pass the result.

…et is used to ensure iteration order is same as insertion order

codelipenghui · 2021-11-03T13:41:51Z

@madhavan-narayanan It should be good to start with a proposal for this new feature, I noticed there are new API introduced. For starting a proposal for Pulsar, you can check https://lists.apache.org/thread/m8dr0hz7qn7rkd48bcp430lcq2q3674g

madhavan-narayanan · 2021-11-03T17:01:35Z

@madhavan-narayanan It should be good to start with a proposal for this new feature, I noticed there are new API introduced. For starting a proposal for Pulsar, you can check https://lists.apache.org/thread/m8dr0hz7qn7rkd48bcp430lcq2q3674g

@codelipenghui , thanks for your comment. I will start a proposal as you suggested.

merlimat · 2021-11-03T16:52:50Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java

@@ -77,6 +77,7 @@
    private int newEntriesCheckDelayInMillis = 10;
    private Clock clock = Clock.systemUTC();
    private ManagedLedgerInterceptor managedLedgerInterceptor;
+    private ManagedLedgerInterceptor managedLedgerPayloadProcessor;


If we're using the same interface, couldn't we have a single managedLedgerInterceptor instance>

merlimat · 2021-11-03T16:53:56Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java

@@ -221,7 +221,7 @@ private void asyncReadEntry0(ReadHandle lh, PositionImpl position, final ReadEnt
                            Iterator<LedgerEntry> iterator = ledgerEntries.iterator();
                            if (iterator.hasNext()) {
                                LedgerEntry ledgerEntry = iterator.next();
-                                EntryImpl returnEntry = EntryImpl.create(ledgerEntry);
+                                EntryImpl returnEntry = EntryImpl.create(ledgerEntry, ml.getConfig().getManagedLedgerPayloadProcessor());


Maybe we can store the interceptor as a member variable of EntryCacheImpl.

merlimat · 2021-11-03T18:43:12Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java

+    public void consumerCreated(ServerCnx cnx,
+                                 Consumer consumer,
+                                 Map<String, String> metadata) {
+        for (BrokerInterceptorWithClassLoader value : interceptors.values()) {


I know this was already the style of other methods here, though we should try to avoid the allocation of an iterator instance in the case where there are no interceptors. eg:

if (interceptors.isEmpty()) { return; } for (BrokerInterceptorWithClassLoader value : interceptors.values()) { // .... }

+1 please check null or empty first.

merlimat · 2021-11-03T18:44:21Z

...r-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerPayloadProcessor.java

+        CompletableFuture<Void> promise = new CompletableFuture<>();
+        promise.complete(null);
+        return promise;


Suggested change

CompletableFuture<Void> promise = new CompletableFuture<>();

promise.complete(null);

return promise;

return CompletableFuture.completedFuture(null);

N/A with the latest changes. This class doesn't exist anymore

merlimat · 2021-11-03T18:48:24Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java

@@ -307,6 +308,7 @@ public TransportCnx getCnx() {
    }

    private static final class MessagePublishContext implements PublishContext, Runnable {
+        Map<String, Object> propertyMap = new HashMap<>();


It's not clear why we need to use the property map here and we should try to avoid the allocation if it's not being used.

Done. propertyMap is created only when needed

merlimat · 2021-11-03T18:49:39Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -1420,11 +1435,15 @@ protected void handleAck(CommandAck ack) {
        final long consumerId = ack.getConsumerId();

        if (consumerFuture != null && consumerFuture.isDone() && !consumerFuture.isCompletedExceptionally()) {
+            Consumer consumer = consumerFuture.getNow(null);


Since we have the consumer variable here, we can also change the next line to avoid doing getNow() again.

merlimat · 2021-11-03T18:51:27Z

pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/PulsarAdminBuilder.java

+     * @param extraClientHeaders
+     * @return
+     */
+    PulsarAdminBuilder extraClientHeaders(String extraClientHeaders);


I think this change should go in a separate PR since it's a bit different from the payload interceptor change.

Also, it's not clear from API perspective how the extra headers are going to be used and what is the format of that string is.

N/A anymore. Removed from the changeset

merlimat · 2021-11-03T18:56:11Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java

+        if(cacheBuf != null && cacheBuf != entry.data)
+            entry.data = cacheBuf;
+        else
+            entry.data.retain();


Instead of checking the equality of the buffer instance, the interceptor should have a well defined semantic on the expectation of the reference count of the returned buffer.

merlimat · 2021-11-03T18:57:03Z

...d-ledger/src/main/java/org/apache/bookkeeper/mledger/intercept/ManagedLedgerInterceptor.java

+     * @param ledgerData data from ledger
+     * @return data after processing, if any
+     */
+    default ByteBuf beforeCacheEntryFromLedger(ByteBuf ledgerData){


I think we need to be very explicit here (and careful :) ) on the reference-count expectations here for the buffers, to avoid memory leaks or use-after-destroyed scenarios.

Agree. The interface is better defined and documented now

merlimat · 2021-11-03T18:58:24Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java

+    }
+
+    @Override
+    public boolean isSuperUser(PulsarService pulsarService, String appId, String originalPrincipal){


I think this would be more suitable to custom authorization provider, where there is already the isSuperUser() method.

Removed from the changeset

madhavan-narayanan · 2021-11-11T10:45:16Z

@madhavan-narayanan It should be good to start with a proposal for this new feature, I noticed there are new API introduced. For starting a proposal for Pulsar, you can check https://lists.apache.org/thread/m8dr0hz7qn7rkd48bcp430lcq2q3674g

@codelipenghui , thanks for your comment. I will start a proposal as you suggested.

Created a feature request #12752

codelipenghui · 2021-11-11T14:54:46Z

@madhavan-narayanan Thanks.

madhavan-narayanan · 2021-12-14T07:21:13Z

@hangc0276 , @codelipenghui , @merlimat , @Jason918
Thanks a lot for your patient review and guidance. This PR will greatly help us as Pulsar cluster operators to move with increased velocity and accelerate internal adoption of Pulsar.

…de cluster better control and flexibility (apache#12536) 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 - 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 - [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 *

…de cluster better control and flexibility (apache#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 *

…de cluster better control and flexibility (apache#12536) 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 - 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 - [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 *

…de cluster better control and flexibility (apache#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 *

…de cluster better control and flexibility (#12536) 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 - 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 - [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 03bbc8e)

…de 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 03bbc8e)

Anonymitaet · 2023-03-24T03:04:23Z

Hi @madhavan-narayanan, thanks for your great contribution!

I see you marked this PR with the doc-required label, but the doc has not been added yet.

To figure out what content we should add, we (+@hangc0276) have gone through the PR changes and figured out the general idea (it adds some inceptors to brokers to better process messages), but we do not know the details.

Since you're the author, can you please add docs or provide technical inputs?

====================

To help you quickly get started, here are some thoughts on doc inputs:

Highly-level structure

How about adding a new topic in Development > Plugin > Interceptor and putting the docs there?

Specific topics

To help devs use this new feature smoothly, consider taking the following doc elements into consideration:
- Who (Developers)
- What (What is this? What features does it provide? → Overview)
- Why (What are the benefits of using this feature? → Use Case)
- When (When to use it? Any considerations or limitations?)
- How (How to use it? → Tutorials (e.g., Get Started) / How-to Guides / Interface Reference doc)
You can take Pulsar Admin API doc as a reference and template. Thank you!

====================

cc @D-2-Ed

…de cluster better control and flexibility (apache#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 *

madhavan-narayanan added 2 commits October 28, 2021 14:49

Broker extensions to allow cluster operators more flexibility and con…

ffea4f7

…trol when operating org-wide messaging platform

Broker extensions to allow cluster operators more flexibility and con…

db58618

…trol when operating org-wide messaging platform

github-actions bot assigned madhavan-narayanan Oct 29, 2021

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Oct 29, 2021

madhavan-narayanan changed the title ~~[pulsar-broker]~~ [pulsar-broker] Broker extensions to allow operators of enterprise wide cluster better control and flexibility Oct 29, 2021

github-actions bot added the doc-required Your PR changes impact docs and you will update later. label Oct 29, 2021

github-actions bot added doc-label-missing and removed doc-not-needed Your PR changes do not impact docs labels Oct 29, 2021

merlimat added the type/enhancement The enhancements for the existing features or docs. e.g. reduce memory usage of the delayed messages label Oct 29, 2021

merlimat added this to the 2.10.0 milestone Oct 29, 2021

github-actions bot removed the doc-label-missing label Oct 29, 2021

Jason918 reviewed Oct 30, 2021

View reviewed changes

To support a list of interceptors for payload processing. LinkedHashS…

1c255de

…et is used to ensure iteration order is same as insertion order

merlimat reviewed Nov 3, 2021

View reviewed changes

Fixes to address review comments

654394e

Client configuration change not needed

5338fee

codelipenghui requested review from 315157973, hangc0276, merlimat, gaoran10, BewareMyPower and congbobo184 December 13, 2021 11:17

codelipenghui merged commit 03bbc8e into apache:master Dec 14, 2021

Anonymitaet mentioned this pull request Dec 31, 2021

[Doc] add docs for "how to develop plugin" in "development" guide #13585

Closed

sijie mentioned this pull request Dec 31, 2021

ISSUE-13585: [Doc] add docs for "how to develop plugin" in "development" guide streamnative/pulsar-archived#3508

Open

RobertIndie mentioned this pull request Aug 9, 2022

[Doc] wrong pip in 2.10.0 release notes #16911

Closed

1 task

congbobo184 added cherry-picked/branch-2.9 Archived: 2.9 is end of life release/2.9.4 labels Nov 17, 2022

poorbarcode mentioned this pull request Feb 2, 2023

[fix] [ml] Fix the incorrect total size if use ML interceptor #19404

Merged

4 tasks

[pulsar-broker] Broker extensions to allow operators of enterprise wide cluster better control and flexibility #12536

[pulsar-broker] Broker extensions to allow operators of enterprise wide cluster better control and flexibility #12536

Conversation

madhavan-narayanan commented Oct 29, 2021 • edited Loading

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

github-actions bot commented Oct 29, 2021

github-actions bot commented Oct 29, 2021

github-actions bot commented Oct 29, 2021

github-actions bot commented Oct 29, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Nov 3, 2021

madhavan-narayanan commented Nov 3, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

madhavan-narayanan commented Nov 11, 2021

codelipenghui commented Nov 11, 2021

madhavan-narayanan commented Dec 14, 2021

Anonymitaet commented Mar 24, 2023

madhavan-narayanan commented Oct 29, 2021 •

edited

Loading