Exposing the broker entry metadata to client #11553

LeBW · 2021-08-04T07:17:02Z

Motivation

This is an improvement to PIP-70. Now the client is able to get the broker entry metadata.

Modifications

increment the client protocol version from 17 to 18.
add a configuration enableExposingBrokerEntryMetadataToClient in the broker configuration.
Let the broker send the broker entry metadata to the client when client protocol version >= 18 and enableExposingBrokerEntryMetadataToClient is true.
Add client support for broker entry metadata. Add two methods getBrokerPublishTime() and getIndex() to utilize the broker entry metadata in Message.

Verifying this change

Make sure that the change passes the CI checks.

Added test for getBrokerPublishTime() and getIndex().

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: (yes)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (yes)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

Documentation

For contributor

For this PR, do we need to update docs?

If yes, please update docs or create a follow-up issue if you need help.

Anonymitaet · 2021-08-05T00:06:25Z

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)

LeBW · 2021-08-05T07:03:07Z

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)

Thanks. "Broker Entry Metadata" is a new feature since 2.8.0, but I haven't found any doc about it for now. Maybe it's better to add some docs for it? (Not just for my PR)

Anonymitaet · 2021-08-05T08:11:53Z

@LeBW thanks for your suggestion. Would you like to add some docs from your perspective?

LeBW · 2021-08-05T08:13:08Z

@LeBW thanks for your suggestion. Would you like to add some docs from your perspective?

Yeah, I can do it later.

Anonymitaet · 2021-08-05T08:26:59Z

@LeBW you can ping me to review the doc and look forward to your contribution, thanks

eolivelli

Great work. I left a few comments

This is an API change and wire protocol change, can you please advertise it on the dev@ mailing list?

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Message.java

LeBW · 2021-08-06T02:48:56Z

Great work. I left a few comments

This is an API change and wire protocol change, can you please advertise it on the dev@ mailing list?

Thanks for your suggestion. I'll write an email to advertise it.

…rom long to Optional<Long>. 2. Improve the unit test.

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

…ryMetadataToClientEnabled

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Message.java

codelipenghui · 2021-08-09T13:09:30Z

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

@@ -186,6 +186,7 @@
    private boolean preciseTopicPublishRateLimitingEnable;
    private boolean encryptionRequireOnProducer;

+    private boolean exposingBrokerEntryMetadataToClientEnabled;


Why do we need to add ServerCnx? Is it able to get from the ServiceConfiguration directly?

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

codelipenghui · 2021-08-09T13:18:52Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java

+
+    @Override
+    public Optional<Long> getIndex() {
+        if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {


For a batch message, we only have one index in the broker entry metadata, we will get the same index for the individual messages of a batch.

Agree with @codelipenghui , for a batch message, we need to caculate the individual index for each message from the Entry index and batch size

Thank you all. Let me fix it.

2. fix getIndex() for batch messages. 3. add E2E tests.

codelipenghui · 2021-08-12T04:50:46Z

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerEntryMetadataE2ETest.java

+        @Cleanup
+        Producer<byte[]> producer = pulsarClient.newProducer()
+                .topic(topic)
+                .enableBatching(true)


The default batching delay is 1ms, it's better to increase the batching delay to such as 1min and force flush the producer, because if we are running the test on a machine which with leak resource, we might get only 1 message in a batch.

I think we can send 5 messages and flush them and send 5 messages again, to flush again, make sure we got 2 batch message and 10 individual messages.

Got it. Fixed.

codelipenghui · 2021-08-12T04:58:51Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java

+        if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {
+            if (msgMetadata.hasNumMessagesInBatch() && messageId instanceof BatchMessageIdImpl) {
+                int batchIndex = ((BatchMessageIdImpl) messageId).getBatchIndex();
+                return Optional.of(brokerEntryMetadata.getIndex() - batchIndex);


If the index is 10 and the batch size is 5, we will get 10, 9, 8, 7, 6 for these 5 messages?

codelipenghui

Great contribution!

codelipenghui · 2021-08-12T10:52:24Z

@eolivelli Please help review the PR again.

eolivelli

Great work !

LGTM

315157973 · 2021-08-12T12:02:50Z

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

+                // skip broker entry metadata if consumer-client doesn't support broker entry metadata or the
+                // features is not enabled
+                if (cnx.getRemoteEndpointProtocolVersion() < ProtocolVersion.v18.getValue()
+                        || !cnx.isExposingBrokerEntryMetadataToClientEnabled()) {


We can get the configuration through the following dependency paths
PulsarCommandSenderImpl -> ServerCnx -> BrokerService -> PulsarService -> configuration
isExposingBrokerEntryMetadataToClientEnabled in ServerCnx can be removed

Thanks for your suggestion. I've removed it in ServerCnx

merlimat · 2021-08-18T01:27:29Z

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

-                Commands.skipBrokerEntryMetadataIfExist(metadataAndPayload);
+                // skip broker entry metadata if consumer-client doesn't support broker entry metadata or the
+                // features is not enabled
+                if (cnx.getRemoteEndpointProtocolVersion() < ProtocolVersion.v18.getValue()


It is better to use the FeatureFlag mechanism instead of checking the ProtocolVersion. This will ensure clients can phase in with the support of this feature.

that's interesting. I wasn't aware of that mechanism

Make sense, even if the client with ProtocolVersion.v18 but does not support get broker entry metadata, the broker can skip the broker entry metadata to reduce the network workload.

It is better to use the FeatureFlag mechanism instead of checking the ProtocolVersion. This will ensure clients can phase in with the support of this feature.

Thanks for your suggestion. Maybe it's better to check both the FeatureFlag and the ProtocolVersion ?

It is better to use the FeatureFlag mechanism instead of checking the ProtocolVersion. This will ensure clients can phase in with the support of this feature.

PIP-70 adds a feature flag supports_broker_entry_metadata, so I use it to check here. Can you review it again?

Anonymitaet · 2021-08-19T00:43:37Z

@LeBW thanks for your contribution! Please do not forget to add docs accordingly to allow users to know your great code changes.

### Motivation This is an improvement to [PIP-70](https://github.com/apache/pulsar/wiki/PIP-70%3A-Introduce-lightweight-broker-entry-metadata). Now the client is able to get the broker entry metadata. ### Modifications 1. increment the client protocol version from 17 to 18. 2. add a configuration `enableExposingBrokerEntryMetadataToClient` in the broker configuration. 3. Let the broker send the broker entry metadata to the client when `client protocol version >= 18` and `enableExposingBrokerEntryMetadataToClient` is true. 4. Add client support for broker entry metadata. Add two methods `getBrokerPublishTime()` and `getIndex()` to utilize the broker entry metadata in `Message`.

This is an improvement to [PIP-70](https://github.com/apache/pulsar/wiki/PIP-70%3A-Introduce-lightweight-broker-entry-metadata). Now the client is able to get the broker entry metadata. 1. increment the client protocol version from 17 to 18. 2. add a configuration `enableExposingBrokerEntryMetadataToClient` in the broker configuration. 3. Let the broker send the broker entry metadata to the client when `client protocol version >= 18` and `enableExposingBrokerEntryMetadataToClient` is true. 4. Add client support for broker entry metadata. Add two methods `getBrokerPublishTime()` and `getIndex()` to utilize the broker entry metadata in `Message`. (cherry picked from commit a7a3721)

Exposing the broker entry metadata to client

a8c45d8

codelipenghui requested a review from hangc0276 August 4, 2021 12:13

codelipenghui assigned LeBW Aug 4, 2021

codelipenghui added this to the 2.9.0 milestone Aug 4, 2021

codelipenghui removed the request for review from hangc0276 August 4, 2021 12:14

Anonymitaet added the doc-required Your PR changes impact docs and you will update later. label Aug 5, 2021

eolivelli requested changes Aug 5, 2021

View reviewed changes

1. modify the return value of "getBrokerPublishTime" and "getIndex" f…

10ac5e7

…rom long to Optional<Long>. 2. Improve the unit test.

aloyszhang reviewed Aug 9, 2021

View reviewed changes

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java Outdated Show resolved Hide resolved

rename enableExposingBrokerEntryMetadataToClient to exposingBrokerEnt…

93f066c

…ryMetadataToClientEnabled

LeBW requested a review from eolivelli August 9, 2021 12:50

LeBW force-pushed the client-brokerentrymetadata branch from 0666188 to 93f066c Compare August 9, 2021 13:00

Merge branch 'master' into client-brokerentrymetadata

325901b

codelipenghui requested changes Aug 9, 2021

View reviewed changes

LeBW added 2 commits August 12, 2021 09:34

1. add two API: haveBrokerPublishTime() and haveIndex()

3f65789

2. fix getIndex() for batch messages. 3. add E2E tests.

Merge branch 'master' into client-brokerentrymetadata

1025378

LeBW force-pushed the client-brokerentrymetadata branch from e5f6a0b to 1025378 Compare August 12, 2021 01:35

codelipenghui reviewed Aug 12, 2021

View reviewed changes

1. fix getIndex() with batch messages. 2. improve test

b6e34e9

codelipenghui approved these changes Aug 12, 2021

View reviewed changes

codelipenghui requested review from hangc0276, 315157973 and merlimat August 12, 2021 10:52

codelipenghui requested review from sijie, BewareMyPower and gaoran10 August 12, 2021 10:52

eolivelli approved these changes Aug 12, 2021

View reviewed changes

315157973 reviewed Aug 12, 2021

View reviewed changes

BewareMyPower approved these changes Aug 12, 2021

View reviewed changes

remove the config in ServerCnx

8512312

LeBW requested a review from 315157973 August 14, 2021 02:54

merlimat reviewed Aug 18, 2021

View reviewed changes

aloyszhang approved these changes Aug 18, 2021

View reviewed changes

add checking feature flag

7fcfb72

LeBW requested a review from merlimat August 19, 2021 01:06

Merge branch 'master' into client-brokerentrymetadata

84f8de6

codelipenghui approved these changes Aug 23, 2021

View reviewed changes

codelipenghui merged commit a7a3721 into apache:master Aug 23, 2021

BewareMyPower mentioned this pull request Sep 29, 2021

[PIP 101] Add seek by index feature for consumer #12234

Closed

Anonymitaet removed the doc-required Your PR changes impact docs and you will update later. label Oct 25, 2021

Anonymitaet mentioned this pull request Oct 25, 2021

[docs][Website] add docs of broker entry metadata #12404

Merged

3 tasks

Anonymitaet mentioned this pull request Dec 15, 2021

[Doc] add explanations for broker entry metadata #13336

Merged

eolivelli added a commit to datastax/starlight-for-kafka that referenced this pull request Feb 9, 2022

add comment about apache/pulsar#11553

6cd77c9

dlg99 mentioned this pull request Jun 15, 2022

Exposing the broker entry metadata to client datastax/pulsar#90

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Exposing the broker entry metadata to client #11553

Exposing the broker entry metadata to client #11553

LeBW commented Aug 4, 2021 •

edited

Loading

Anonymitaet commented Aug 5, 2021

LeBW commented Aug 5, 2021

Anonymitaet commented Aug 5, 2021

LeBW commented Aug 5, 2021

Anonymitaet commented Aug 5, 2021 •

edited

Loading

eolivelli left a comment

LeBW commented Aug 6, 2021

codelipenghui Aug 9, 2021

codelipenghui Aug 9, 2021

aloyszhang Aug 10, 2021 •

edited

Loading

LeBW Aug 10, 2021

LeBW Aug 11, 2021

codelipenghui Aug 12, 2021

codelipenghui Aug 12, 2021

LeBW Aug 12, 2021

codelipenghui Aug 12, 2021

LeBW Aug 12, 2021

codelipenghui left a comment

codelipenghui commented Aug 12, 2021

eolivelli left a comment

315157973 Aug 12, 2021

LeBW Aug 12, 2021

merlimat Aug 18, 2021

eolivelli Aug 18, 2021

codelipenghui Aug 18, 2021

LeBW Aug 18, 2021 •

edited

Loading

LeBW Aug 19, 2021

Anonymitaet commented Aug 19, 2021

Exposing the broker entry metadata to client #11553

Exposing the broker entry metadata to client #11553

Conversation

LeBW commented Aug 4, 2021 • edited Loading

Motivation

Modifications

Verifying this change

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

Documentation

For contributor

Anonymitaet commented Aug 5, 2021

LeBW commented Aug 5, 2021

Anonymitaet commented Aug 5, 2021

LeBW commented Aug 5, 2021

Anonymitaet commented Aug 5, 2021 • edited Loading

eolivelli left a comment

Choose a reason for hiding this comment

LeBW commented Aug 6, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aloyszhang Aug 10, 2021 • edited Loading

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

codelipenghui left a comment

Choose a reason for hiding this comment

codelipenghui commented Aug 12, 2021

eolivelli left a 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

LeBW Aug 18, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Anonymitaet commented Aug 19, 2021

LeBW commented Aug 4, 2021 •

edited

Loading

Anonymitaet commented Aug 5, 2021 •

edited

Loading

aloyszhang Aug 10, 2021 •

edited

Loading

LeBW Aug 18, 2021 •

edited

Loading