Broker side deduplication #751

merlimat · 2017-09-08T03:58:03Z

Motivation

Context in PIP-6.

This PR contains the broker side changes to implement the deduplication using the logic described in the design doc.

This is not the complete feature. I will add unit tests after pushing the client library changes as well. All tasks are in the deduplication project

Modifications

Added APIs to enable/disable deduplication
Check for sequence ids to discard duplicates
Use the cursor properties to take snapshot of the sequence id map
Load the snapshot and replay the cursor when loading the topic

merlimat · 2017-09-08T03:58:28Z

cc/ @sijie @zhaijack. Please take a look as well

rdhabalia · 2017-09-09T00:16:51Z

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

+    }
+
+    public PersistentTopic(String topic, ManagedLedger ledger, BrokerService brokerService,
+            PersistentTopicConfiguration topicConfiguration) {


is there any reason of having PersistentTopicConfiguration?

explained here below.

rdhabalia · 2017-09-09T00:41:16Z

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

                    new OpenLedgerCallback() {
                        @Override
                        public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
-                            PersistentTopic persistentTopic = new PersistentTopic(topic, ledger, BrokerService.this);
+                            PersistentTopic persistentTopic = new PersistentTopic(topic, ledger, BrokerService.this,
+                                    config);


I don't see PersistentTopicConfiguration being used in PersistentTopic. In that case, we don't need PersistentTopicConfiguration ?

Before this PR, we were getting the per-topic (actually it's per-namespace) configuration only for ManagedLedger purposes: https://github.com/apache/incubator-pulsar/blob/aa40ebbe02f9b4f60b7f3910c293c50c18f35687/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java#L567

For deduplication, when creating the topic I need to know wether it's enabled or not, depending on the broker default and on the per-namespace policy.

PersistentTopicConfiguration is grouping both the policy and the customized ManagedLedgerConfig.

For deduplication, when creating the topic I need to know wether it's enabled or not
PersistentTopicConfiguration is grouping both the policy and the customized ManagedLedgerConfig.

Yes, I understood that part. But correct if I am wrong:
MessageDeduplication checks namespace-dedup by directly fetching from policies-cache and PersistentTopicConfiguration.namespacePolicies is not being used anywhere.
so, I don't see any difference if we just return ManagedLedgerConfig instead PersistentTopicConfiguration because it seems we are not using it anywhere?

Uhm, I guess you're right :)

I have indeed changed the approach after adding the PersistentTopicConfiguration so that isDeduplicationEnabled() is checking the policy and returning a future.

I'll remove PersistentTopicConfiguration

I'll remove PersistentTopicConfiguration

Should we revert getTopicConfiguration() to getManagedLedgerConfig() if we are not using PersistentTopicConfiguration?

Ouch, I finally forgot to remove it. Will do that

rdhabalia · 2017-09-09T00:47:16Z

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

+                    highestSequencedPersisted.put(producerName, sequenceId);
+
+                    md.recycle();
+                    messageMetadataAndPayload.release();


should we release entry as well?

either we release the entry or the buffer messageMetadataAndPayload, the effect is the same, though we cannot release both.

either we release the entry or the buffer messageMetadataAndPayload, the effect is the same

Releasing Entry also releases payload(messageMetadataAndPayload) but just releasing payload will not release Entry so, can you correct if it it's wrong but here we are not releasing Entry so, isn't that leak?

I'll change it to call entry.release(). In any case it won't be a "leak", just a missed opportunity to recycling the entry instance.

In any case it won't be a "leak"

Yes, but if we don't recycle EntryImpl then don't we see leak with netty-detection report and it will keep creating new objects and will go out of memory.?

No no, it's not that bad. When "leaking" objects from a recycler, they will just get normally GCed.

rdhabalia · 2017-09-09T00:50:00Z

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

+
+                    String producerName = md.getProducerName();
+                    long sequenceId = md.getSequenceId();
+                    highestSequencedPushed.put(producerName, sequenceId);


what if we have dedup disabled first and then enabled/ Or old-client lib published msg with null producer-name. In that case, some of the metadata will have producerName=null. So, ConcurrentOpenHashMap.put(null,x) will throw RuntimeException?

Producer name was present in the metadata (and it's required) since the very beginning of Pulsar. It will never be null.

rdhabalia · 2017-09-09T01:06:40Z

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

+     * returning a future to track the completion of the task
+     */
+    public CompletableFuture<Void> checkStatus() {
+        if (status == Status.Recovering || status == Status.Removing) {


can we move this check at ln#197 after getting isDeduplicationEnabled() because sometimes, we have seen (mostly while removing replicator cluster from policies) that broker receives multiple zk-watch event and we have to handle this race condition. Here, isDeduplicationEnabled() is an async therefore, it make sense to move this check once we get the result from it (if possible then under synchronized block).

Makes sense, also I'll add a mutex in the state mutation

rdhabalia · 2017-09-09T01:08:16Z

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

+                    public void deleteCursorFailed(ManagedLedgerException exception, Object ctx) {
+                        log.warn("[{}] Failed to disable deduplication: {}", topic.getName(), exception.getMessage());
+                        status = Status.Failed;
+                        future.completeExceptionally(exception);


is it a ledger leak? should we retry X time?

It's not a ledger leak since it's the delete cursor operation that failed. In this case the cursor will stay there. I'll make sure to schedule a check in this case to retry to disable deduplication.

rdhabalia · 2017-09-09T01:13:17Z

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

+                        }).exceptionally(ex -> {
+                            status = Status.Failed;
+                            log.warn("[{}] Failed to enable deduplication: {}", topic.getName(), ex.getMessage());
+                            future.completeExceptionally(ex);


I think status should be part of client metrics to verify that dedup is not enabled.

You mean, reported in the topic stats ?

yes, it would be useful to know in case status=failed. so, does it make sense to add in topic-stats?

rdhabalia · 2017-09-09T01:19:16Z

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

+                return true;
+            }
+
+            highestSequencedPushed.put(producerName, sequenceId);


I feel isMessageAlreadyStored name is little confusing.. as per method name it should not update the map. I know we want to add the sequenceId as well. how about pushSequenceId() and if it returns false means couldn't store.

Ok, let me rethink the method name.

Renaming into shouldPublishNextMessage()

merlimat · 2017-09-12T05:11:45Z

@rdhabalia Updated addressing comments. Also added logic to check for original producer name in case of replicated messages.

rdhabalia

LGTM.. just a minor question regarding use of PersistentTopicConfiguration

sijie

LGTM +1

rdhabalia

👍

rdhabalia · 2017-10-31T01:52:31Z

I think we don't have a testcase for dedup..right?

merlimat · 2017-10-31T02:40:20Z

@rdhabalia There's a test class at https://github.com/apache/incubator-pulsar/blob/d4a8c25bc00bc1d61e1a1f5b4ee34566cdacfe3d/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java

Broker side deduplication

4e1e522

merlimat added the type/feature The PR added a new feature or issue requested a new feature label Sep 8, 2017

merlimat added this to the 1.20.0-incubating milestone Sep 8, 2017

merlimat self-assigned this Sep 8, 2017

merlimat requested review from rdhabalia, jai1 and saandrews September 8, 2017 03:58

merlimat added this to In progress in Message deduplication Sep 8, 2017

merlimat moved this from In progress to PR available in Message deduplication Sep 8, 2017

rdhabalia reviewed Sep 9, 2017

View reviewed changes

Addressed comments

07d7f94

rdhabalia reviewed Sep 12, 2017

View reviewed changes

Removed PersistentTopicConfiguration

c15aad6

sijie approved these changes Sep 12, 2017

View reviewed changes

rdhabalia approved these changes Sep 12, 2017

View reviewed changes

merlimat merged commit bdbb121 into apache:master Sep 13, 2017

merlimat moved this from PR available to Merged in Message deduplication Sep 13, 2017

merlimat deleted the broker-side-deduplication branch October 31, 2017 02:38

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Broker side deduplication #751

Broker side deduplication #751

merlimat commented Sep 8, 2017

merlimat commented Sep 8, 2017

rdhabalia Sep 9, 2017

merlimat Sep 9, 2017

rdhabalia Sep 9, 2017

merlimat Sep 9, 2017

rdhabalia Sep 11, 2017

merlimat Sep 11, 2017

rdhabalia Sep 12, 2017

merlimat Sep 12, 2017

rdhabalia Sep 9, 2017

merlimat Sep 9, 2017

rdhabalia Sep 11, 2017

merlimat Sep 11, 2017

rdhabalia Sep 11, 2017

merlimat Sep 12, 2017

rdhabalia Sep 9, 2017

merlimat Sep 9, 2017

rdhabalia Sep 9, 2017

merlimat Sep 12, 2017

rdhabalia Sep 9, 2017

merlimat Sep 9, 2017

rdhabalia Sep 9, 2017

merlimat Sep 9, 2017

rdhabalia Sep 11, 2017

merlimat Sep 12, 2017

rdhabalia Sep 9, 2017

merlimat Sep 9, 2017

merlimat Sep 12, 2017

merlimat commented Sep 12, 2017

rdhabalia left a comment

sijie left a comment

rdhabalia left a comment

rdhabalia commented Oct 31, 2017

merlimat commented Oct 31, 2017

Broker side deduplication #751

Broker side deduplication #751

Conversation

merlimat commented Sep 8, 2017

Motivation

Modifications

merlimat commented Sep 8, 2017

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

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

merlimat commented Sep 12, 2017

rdhabalia left a comment

Choose a reason for hiding this comment

sijie left a comment

Choose a reason for hiding this comment

rdhabalia left a comment

Choose a reason for hiding this comment

rdhabalia commented Oct 31, 2017

merlimat commented Oct 31, 2017