[PIP 70][Issue 8617] Introduce lightweight broker entry metadata #8618

aloyszhang · 2020-11-18T18:00:05Z

Fixes #8617

Motivation

Introduce lightweight raw Message metadata, details can be found PIP-70

Modifications

wire protocol add RawMessageMetadata and supports_raw_message_meta for FeatureFlags
change how produced message is saved in bookkeeper: add raw metadata for message
change how message is seek-by-time
change how message send back to Consumer: skip metadata if consumer not supprot raw metadata

Verifying this change

Added tests for parse/skip raw message metadata
Added test for how message seek-by broker timestamp for message

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

If yes was chosen, please highlight the changes

The wire protocol: (yes / no)

aloyszhang · 2020-11-19T06:05:53Z

/pulsarbot run-failure-checks

aloyszhang · 2020-11-19T08:58:40Z

/pulsarbot run-failure-checks

aloyszhang · 2020-11-19T10:57:55Z

/pulsarbot run-failure-checks

jiazhai · 2020-11-21T02:37:27Z

@BewareMyPower Would you please also take a look?

BewareMyPower · 2020-11-21T08:46:54Z

@jiazhai Ok

BewareMyPower

I think we should add a method isRawMetadataEnable() to ManagedLedgerConfig like:

public boolean isRawMetadataEnable() {
    return isBrokerTimestampForMessageEnable();
}

Then if we want to support more features like message sequence id, we'll only need to change this method to return isConfig1() || isConfig2() || .... And other occurrences like OpAndEntry#initiate() need no changes.

aloyszhang · 2020-11-21T14:57:41Z

@BewareMyPower Would you give some deatils about what the method ManagedLedgerConfig#isRawMetadataEnable() is used for?

BewareMyPower · 2020-11-21T15:17:48Z

See

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

Lines 1360 to 1362 in 0c71eb6

    
           if (config.isBrokerTimestampForMessageEnable()) { 
        
               existsOp = OpAddEntry.create(existsOp.ml, existsOp.dataWithRawMetadata, existsOp.callback, existsOp.ctx); 
        
           } else {

and

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

Lines 112 to 114 in 0c71eb6

    
           if (ml.getConfig().isBrokerTimestampForMessageEnable()) { 
        
               duplicateBuffer = Commands.addRawMessageMetadata(duplicateBuffer); 
        
               dataWithRawMetadata = duplicateBuffer.retainedDuplicate();

These two code blocks both check isBrokerTimestampForMessageEnable and determine whether to serialize or deserialize RawMetadata. However, if we add a new field to RawMessageMetadata, e.g. message_sequence_id, the check needs to be changed to

if (config.isBrokerTimestampForMessageEnable() || config.isMessageSequenceIdEnable())

Right? So we should wrap the check into a single method, like:

public boolean isRawMetadataEnable() {
    return isBrokerTimestampForMessageEnable() /* || isMessageSequenceIdEnable() or more checks */;
}

aloyszhang · 2020-11-22T01:29:11Z

@BewareMyPower
Thanks for your suggention.
I'll add ManagedLedgerConfig#isRawMetadataEnable() before serialize/deserialize.

aloyszhang · 2020-11-23T08:22:47Z

/pulsarbot run-failure-checks

aloyszhang · 2020-11-23T08:49:24Z

@BewareMyPower I applied your comment, PTAL.

BewareMyPower

LGTM

aloyszhang · 2020-11-23T09:23:36Z

/pulsarbot run-failure-checks

aloyszhang · 2020-11-23T10:35:22Z

/pulsarbot run-failure-checks

aloyszhang · 2020-11-23T11:40:43Z

/pulsarbot run-failure-checks

jiazhai · 2020-11-24T02:22:46Z

/pulsarbot run-failure-checks

aloyszhang · 2020-11-24T07:50:34Z

/pulsarbot run-failure-checks

jiazhai · 2020-11-24T10:44:25Z

@aloyszhang Thanks for the great work. Would you please help add backward compact test, there was already some tests, e.g. SmokeTest2_5 to do this, maybe we need only change the broker.conf to enable this feature, and test the old client could read data correctly.

aloyszhang · 2020-11-25T02:41:49Z

@jiazhai I will hanlde backward compact test soon.

jiazhai · 2020-12-01T09:44:34Z

/pulsarbot run-failure-checks

jiazhai · 2020-12-02T01:25:06Z

/pulsarbot run-failure-checks

…adata

aloyszhang · 2020-12-10T15:49:30Z

/pulsarbot run-failure-checks

sijie · 2020-12-10T18:07:11Z

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

@@ -865,6 +866,11 @@
                "please enable the system topic first.")
    private boolean topicLevelPoliciesEnabled = false;

+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "List of interceptors for broker metadata.")


Suggested change

doc = "List of interceptors for broker metadata.")

doc = "List of interceptors for entry metadata.")

will fix this.

sijie · 2020-12-10T18:16:42Z

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


+            ByteBuf duplicateBuffer = data.retainedDuplicate();
+            if (ml.getConfig().isBrokerEntryMetaEnabled()) {
+                duplicateBuffer = Commands.addBrokerEntryMetadata(duplicateBuffer,


I am still not convinced why do we need to this here. ManagedLedger only handles serialized entry. The entry metadata should be appended at the broker level. I think the right place to add this logic should be done in https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L342.

As we described in PIP-70, this feature can support continous sequenceId for Pulsar entry in the future, so, I think do these operations here is in favour of the future features. For this broker-timestamp feature only, move this logic as you suggested is a good choice. But for further extension, maybe do this logic by ManagedLedger is better. What's your opinion about this?

sijie · 2020-12-10T18:22:59Z

...rc/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java

                try {
-                    msg = MessageImpl.deserialize(entry.getDataBuffer());
-                    return msg.isExpired(messageTTLInSeconds);
+                    pair = MessageImpl.deserializeWithBrokerEntryMetaData(entry.getDataBuffer());


We should expose the broker metadata in the Message. So this would avoid using Pair and a lot of if-else logic.

We can improve msg.isExpired logic. If entry metadata is present, use broker timestamp; otherwise use client timestamp.

Thanks for your suggestion. I'll optimize the implement as follow steps:

expose the entry metadata to MessageImpl

if entry metadata exist, skip the deserialization of MessageMetadata

if not exist, deserialization the MessageMetadata

…try meta

aloyszhang · 2020-12-11T07:13:29Z

/pulsarbot run-failure-checks

…f comparation condition

aloyszhang · 2020-12-11T10:09:44Z

/pulsarbot run-failure-checks

sijie

@aloyszhang the change looks great now! I left a couple of comments. Once you address them, I think this change is ready to merge.

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

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

...rc/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java

...roker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageFinder.java

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

...c/main/java/org/apache/pulsar/common/intercept/AppendBrokerTimestampMetadataInterceptor.java

pulsar-common/src/main/java/org/apache/pulsar/common/intercept/BrokerEntryMetadataUtils.java

pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java

sijie · 2020-12-12T03:13:09Z

@aloyszhang great job!

Anonymitaet · 2021-05-12T08:10:19Z

@aloyszhang thanks for your great work.

Shall the changes be documented in the user guide?

If so, could you please help add the docs accordingly? Then you can ping me to review, thanks

aloyszhang force-pushed the raw-meta branch from 276960e to 0c71eb6 Compare November 19, 2020 03:59

jiazhai changed the title ~~[Issue 8617] Introduce lightweight raw Message metadata~~ [PIP 70][Issue 8617] Introduce lightweight raw Message metadata Nov 19, 2020

jiazhai added the type/PIP label Nov 19, 2020

jiazhai requested review from codelipenghui, sijie, jerrypeng and merlimat November 21, 2020 02:37

BewareMyPower reviewed Nov 21, 2020

View reviewed changes

BewareMyPower approved these changes Nov 23, 2020

View reviewed changes

aloyszhang force-pushed the raw-meta branch from 28c7953 to a07504d Compare November 27, 2020 04:31

aloyszhang force-pushed the raw-meta branch from 0054cc7 to f329ac5 Compare December 2, 2020 03:42

aloyszhang added 3 commits December 10, 2020 21:14

skip raw meta when skip medtadata

6af2d75

skip raw metadata before send to consumer

a91c771

introduce broker entry metadta interceptor for adding broker side met…

397939b

…adata

aloyszhang force-pushed the raw-meta branch from bdff5ab to 397939b Compare December 10, 2020 13:14

fix checkstyle

4e64b96

sijie requested changes Dec 10, 2020

View reviewed changes

sijie changed the title ~~[PIP 70][Issue 8617] Introduce lightweight raw Message metadata~~ [PIP 70][Issue 8617] Introduce lightweight broker entry metadata Dec 10, 2020

aloyszhang added 5 commits December 11, 2020 10:32

fix config doc

6ea94a5

expose entry meta to MessageImpl and skip deser MessageMeta if has en…

48c0b78

…try meta

move add entry metadta logic to topic

e1a981b

fix checkstyle

931ea7c

fix checkstyle

0f2a6e8

codelipenghui assigned aloyszhang Dec 11, 2020

codelipenghui added this to the 2.8.0 milestone Dec 11, 2020

fix error in PulsarSplitManager.findPosition by roll back the logic o…

6e2d971

…f comparation condition

jiazhai approved these changes Dec 11, 2020

View reviewed changes

sijie reviewed Dec 11, 2020

View reviewed changes

aloyszhang added 5 commits December 12, 2020 06:53

remove un-used import and method

23c9c2c

make all time comparation happended in Messge

43556cb

throw runtime exception if load BrokerEntryMetadataInterceptor failed

76b6f3f

catch add entry metadata exception and failed write request

1e918b5

remove un-used import

5e237c4

sijie approved these changes Dec 12, 2020

View reviewed changes

jiazhai merged commit 6275297 into apache:master Dec 12, 2020

aloyszhang deleted the raw-meta branch December 25, 2020 03:07

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[PIP 70][Issue 8617] Introduce lightweight broker entry metadata #8618

[PIP 70][Issue 8617] Introduce lightweight broker entry metadata #8618

aloyszhang commented Nov 18, 2020

aloyszhang commented Nov 19, 2020

aloyszhang commented Nov 19, 2020

aloyszhang commented Nov 19, 2020

jiazhai commented Nov 21, 2020

BewareMyPower commented Nov 21, 2020

BewareMyPower left a comment

aloyszhang commented Nov 21, 2020

BewareMyPower commented Nov 21, 2020

aloyszhang commented Nov 22, 2020

aloyszhang commented Nov 23, 2020

aloyszhang commented Nov 23, 2020

BewareMyPower left a comment

aloyszhang commented Nov 23, 2020

aloyszhang commented Nov 23, 2020

aloyszhang commented Nov 23, 2020

jiazhai commented Nov 24, 2020

aloyszhang commented Nov 24, 2020

jiazhai commented Nov 24, 2020

aloyszhang commented Nov 25, 2020 •

edited

Loading

jiazhai commented Dec 1, 2020

jiazhai commented Dec 2, 2020

aloyszhang commented Dec 10, 2020

sijie Dec 10, 2020

aloyszhang Dec 11, 2020

sijie Dec 10, 2020

aloyszhang Dec 11, 2020

sijie Dec 10, 2020

aloyszhang Dec 11, 2020 •

edited

Loading

aloyszhang commented Dec 11, 2020

aloyszhang commented Dec 11, 2020

sijie left a comment

sijie commented Dec 12, 2020

Anonymitaet commented May 12, 2021

	doc = "List of interceptors for broker metadata.")
	doc = "List of interceptors for entry metadata.")

[PIP 70][Issue 8617] Introduce lightweight broker entry metadata #8618

[PIP 70][Issue 8617] Introduce lightweight broker entry metadata #8618

Conversation

aloyszhang commented Nov 18, 2020

Motivation

Modifications

Verifying this change

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

aloyszhang commented Nov 19, 2020

aloyszhang commented Nov 19, 2020

aloyszhang commented Nov 19, 2020

jiazhai commented Nov 21, 2020

BewareMyPower commented Nov 21, 2020

BewareMyPower left a comment

Choose a reason for hiding this comment

aloyszhang commented Nov 21, 2020

BewareMyPower commented Nov 21, 2020

aloyszhang commented Nov 22, 2020

aloyszhang commented Nov 23, 2020

aloyszhang commented Nov 23, 2020

BewareMyPower left a comment

Choose a reason for hiding this comment

aloyszhang commented Nov 23, 2020

aloyszhang commented Nov 23, 2020

aloyszhang commented Nov 23, 2020

jiazhai commented Nov 24, 2020

aloyszhang commented Nov 24, 2020

jiazhai commented Nov 24, 2020

aloyszhang commented Nov 25, 2020 • edited Loading

jiazhai commented Dec 1, 2020

jiazhai commented Dec 2, 2020

aloyszhang commented Dec 10, 2020

sijie Dec 10, 2020

Choose a reason for hiding this comment

aloyszhang Dec 11, 2020

Choose a reason for hiding this comment

sijie Dec 10, 2020

Choose a reason for hiding this comment

aloyszhang Dec 11, 2020

Choose a reason for hiding this comment

sijie Dec 10, 2020

Choose a reason for hiding this comment

aloyszhang Dec 11, 2020 • edited Loading

Choose a reason for hiding this comment

aloyszhang commented Dec 11, 2020

aloyszhang commented Dec 11, 2020

sijie left a comment

Choose a reason for hiding this comment

sijie commented Dec 12, 2020

Anonymitaet commented May 12, 2021

aloyszhang commented Nov 25, 2020 •

edited

Loading

aloyszhang Dec 11, 2020 •

edited

Loading