[feat][broker][PIP-195] Implement delayed message index bucket snapshot(merge/delete) - part8 #19138

coderzc · 2023-01-05T11:03:25Z

Motivation

Modifications

Add DelayedIndexQueue interface to abstract CombinedSegmentDelayedIndexQueue and TripleLongPriorityDelayedIndexQueue.

Implement delayed message index bucket snapshot(merge/delete).

Verifying this change

Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

testMergeSnapshot / DelayedIndexQueueTest

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository: coderzc#35

poorbarcode

After BucketDelayedDeliveryTracker.recover, part of the delayed messages will be put into the delayed message queue, but these messages may be reread by the managed cursor. Will this cause repeated delivery?

poorbarcode · 2023-01-11T02:57:19Z

...c/main/java/org/apache/pulsar/broker/delayed/bucket/TripleLongPriorityDelayedIndexQueue.java

+import org.apache.pulsar.common.util.collections.TripleLongPriorityQueue;
+
+@NotThreadSafe
+public class TripleLongPriorityDelayedIndexQueue implements DelayedIndexQueue {


I notice that this class is used to make peek and pop easier to use, and it is only used by MutableBucket, should it not exist as a separate public class?

I will remove public modifier.

poorbarcode · 2023-01-11T02:58:59Z

.../src/main/java/org/apache/pulsar/broker/delayed/bucket/CombinedSegmentDelayedIndexQueue.java

+import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat.SnapshotSegment;
+
+@NotThreadSafe
+public class CombinedSegmentDelayedIndexQueue implements DelayedIndexQueue {


I notice that this class is used to make peek and pop easier to use, and it is only used by merging two queues in BucketDelayedDeliveryTracker, should it not exist as a separate public class?

Hello @coderzc

And this comment (^_^)

Sorry, I missed this

poorbarcode · 2023-01-11T03:00:40Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/DelayedIndexQueue.java

+import java.util.Objects;
+import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat;
+
+public interface DelayedIndexQueue {


This interface is only used by MutableBucket and is not universal. Is this interface not needed？

I will remove public modifier.

poorbarcode · 2023-01-11T03:04:40Z

...oker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java

+                        CompletableFuture<Void> removeAFuture = bucketA.asyncDeleteBucketSnapshot();
+                        CompletableFuture<Void> removeBFuture = bucketB.asyncDeleteBucketSnapshot();
+
+                        return CompletableFuture.allOf(removeAFuture, removeBFuture).thenRun(() -> {


We do save the newly merged bucket after finished delete the original buckets, it may be caused bucket loss. I think losing buckets is tolerable, right?

Maybe we can persist the newly merged bucket, then delete the original buckets. If failed to save the newly merged bucket, the tracker needs to rebuild the index, right?

If the bucket index is lost, the tracker can rebuild the index, but if there are two overlapped buckets, it will cause data confusion.

Delete the original bucket index first, then save the newly merged index, this may lose the bucket index, but the tracker will rebuild the index.

Save the newly merged index first, then delete the original bucket index, this may cause intersecting buckets, and we need to find a way to clean up the original buckets.

OK, I think there are two ways, the first way is simple and clear.

Questions:

When is the index rebuilt?

Should we unload the topic to rebuild the index if this operation fails?

When is the index rebuilt?

If index is lost, the tracker will rebuild index when reread by the managed cursor.

Should we unload the topic to rebuild the index if this operation fails?

Yes, we can unload the topic to rebuild the index

Can we mark the old segment as deleting or merging? So that we can continue the merge operation after the broker crashes.

Can we mark the old segment as deleting or merging? So that we can continue the merge operation after the broker crashes.

This will introduce more state in metadata, we can improve it when we really need it

I find a way to clean up the old overlap buckets when recovering buckets, so that we can persist the newly merged bucket first, then delete the original bucket.

The approach is as follows:
If there is range [1...10]、[30...40]、[1...40], then it will only remain [1...40]

.../src/main/java/org/apache/pulsar/broker/delayed/bucket/CombinedSegmentDelayedIndexQueue.java

coderzc · 2023-01-11T08:27:38Z

After BucketDelayedDeliveryTracker.recover, part of the delayed messages will be put into the delayed message queue, but these messages may be reread by the managed cursor. Will this cause repeated delivery?

These messages that have already been added tracker will be skipped when reread by the managed cursor, More see: #19035

gaoran10

Nice work! LGTM

mattisonchao · 2023-01-13T02:24:23Z

...oker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java

-                // TODO merge bucket snapshot (synchronize operate)
+                try {
+                    asyncMergeBucketSnapshot().get(AsyncOperationTimeoutSeconds, TimeUnit.SECONDS);
+                } catch (InterruptedException | ExecutionException | TimeoutException e) {


Please call Thread.currentThread().interrupt(); if we get InterruptedException.

Hi @mattisonchao
I wonder why we should keep the interrupted state. Is this state to be handed over to netty?

Maybe not. But I think we should persist in this state if we don't have any special purpose.

mattisonchao · 2023-01-13T02:24:56Z

...oker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java

+                try {
+                    asyncMergeBucketSnapshot().get(AsyncOperationTimeoutSeconds, TimeUnit.SECONDS);
+                } catch (InterruptedException | ExecutionException | TimeoutException e) {
+                    throw new RuntimeException(e);


Do you think we should consider the specific exception?

pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java

mattisonchao · 2023-01-13T02:37:12Z

...oker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java

+                            synchronized (immutableBuckets) {
+                                immutableBuckets.remove(Range.closed(bucket.startLedgerId, bucket.endLedgerId));
+                            }
+                            bucket.asyncDeleteBucketSnapshot();


We need a log to show the exception.

Add exception log in asyncDeleteBucketSnapshot

...oker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java

mattisonchao

Overall LGTM, I need to review the whole logic after this PR to ensure I got all the context.

codecov-commenter · 2023-01-13T13:00:55Z

Codecov Report

Merging #19138 (eb33c21) into master (4b8f447) will decrease coverage by 0.69%.
The diff coverage is 0.00%.

@@             Coverage Diff              @@
##             master   #19138      +/-   ##
============================================
- Coverage     47.00%   46.30%   -0.70%     
- Complexity    10639    18434    +7795     
============================================
  Files           713     1625     +912     
  Lines         69672   132248   +62576     
  Branches       7482    14560    +7078     
============================================
+ Hits          32746    61234   +28488     
- Misses        33212    64735   +31523     
- Partials       3714     6279    +2565

Flag	Coverage Δ
unittests	`46.30% <0.00%> (-0.70%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...rg/apache/pulsar/broker/delayed/bucket/Bucket.java	`0.00% <0.00%> (ø)`
...r/delayed/bucket/BucketDelayedDeliveryTracker.java	`0.00% <0.00%> (ø)`
...layed/bucket/CombinedSegmentDelayedIndexQueue.java	`0.00% <0.00%> (ø)`
...ulsar/broker/delayed/bucket/DelayedIndexQueue.java	`0.00% <0.00%> (ø)`
.../pulsar/broker/delayed/bucket/ImmutableBucket.java	`0.00% <0.00%> (ø)`
...he/pulsar/broker/delayed/bucket/MutableBucket.java	`0.00% <0.00%> (ø)`
...ed/bucket/TripleLongPriorityDelayedIndexQueue.java	`0.00% <0.00%> (ø)`
...pache/pulsar/client/impl/CompactionReaderImpl.java	`0.00% <0.00%> (-90.00%)`	⬇️
...g/apache/pulsar/client/impl/RawBatchConverter.java	`3.12% <0.00%> (-89.07%)`	⬇️
.../pulsar/compaction/StrategicTwoPhaseCompactor.java	`0.00% <0.00%> (-76.20%)`	⬇️
... and 1066 more

coderzc added 3 commits January 5, 2023 15:27

Implement bucket snapshot merge/delete

9861ea3

fix license

c27a194

fix ConcurrentModificationException

3d96a69

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Jan 5, 2023

coderzc self-assigned this Jan 5, 2023

coderzc added this to the 2.12.0 milestone Jan 5, 2023

coderzc added type/feature The PR added a new feature or issue requested a new feature area/broker labels Jan 5, 2023

coderzc requested review from mattisonchao and codelipenghui January 5, 2023 11:10

poorbarcode reviewed Jan 11, 2023

View reviewed changes

gaoran10 reviewed Jan 11, 2023

View reviewed changes

.../src/main/java/org/apache/pulsar/broker/delayed/bucket/CombinedSegmentDelayedIndexQueue.java Outdated Show resolved Hide resolved

Address comment

621a8e8

coderzc force-pushed the bucket_delayed_merge_delete branch from aa805a1 to 621a8e8 Compare January 11, 2023 12:34

gaoran10 reviewed Jan 11, 2023

View reviewed changes

gaoran10 approved these changes Jan 11, 2023

View reviewed changes

mattisonchao reviewed Jan 13, 2023

View reviewed changes

pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java Outdated Show resolved Hide resolved

mattisonchao reviewed Jan 13, 2023

View reviewed changes

...oker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java Show resolved Hide resolved

mattisonchao approved these changes Jan 13, 2023

View reviewed changes

coderzc force-pushed the bucket_delayed_merge_delete branch from 76a8619 to 525e30f Compare January 13, 2023 11:09

Address comment

14b5ed2

coderzc force-pushed the bucket_delayed_merge_delete branch from 525e30f to 14b5ed2 Compare January 13, 2023 11:10

remove public modifier

90c6690

poorbarcode approved these changes Jan 16, 2023

View reviewed changes

clean up old overlap buckets when recover buckets

eb33c21

codelipenghui merged commit 9dc490f into apache:master Jan 17, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[feat][broker][PIP-195] Implement delayed message index bucket snapshot(merge/delete) - part8 #19138

[feat][broker][PIP-195] Implement delayed message index bucket snapshot(merge/delete) - part8 #19138

coderzc commented Jan 5, 2023

poorbarcode left a comment

poorbarcode Jan 11, 2023

coderzc Jan 11, 2023

poorbarcode Jan 11, 2023

poorbarcode Jan 15, 2023

coderzc Jan 16, 2023

poorbarcode Jan 11, 2023

coderzc Jan 11, 2023

poorbarcode Jan 11, 2023

gaoran10 Jan 11, 2023 •

edited

Loading

coderzc Jan 11, 2023 •

edited

Loading

gaoran10 Jan 11, 2023 •

edited

Loading

mattisonchao Jan 13, 2023 •

edited

Loading

coderzc Jan 13, 2023

codelipenghui Jan 16, 2023

coderzc Jan 16, 2023

coderzc Jan 16, 2023 •

edited

Loading

coderzc commented Jan 11, 2023 •

edited

Loading

gaoran10 left a comment

mattisonchao Jan 13, 2023

poorbarcode Jan 13, 2023

mattisonchao Jan 13, 2023

mattisonchao Jan 13, 2023

mattisonchao Jan 13, 2023

coderzc Jan 13, 2023

mattisonchao left a comment •

edited

Loading

codecov-commenter commented Jan 13, 2023 •

edited

Loading

[feat][broker][PIP-195] Implement delayed message index bucket snapshot(merge/delete) - part8 #19138

[feat][broker][PIP-195] Implement delayed message index bucket snapshot(merge/delete) - part8 #19138

Conversation

coderzc commented Jan 5, 2023

Motivation

Modifications

Verifying this change

Documentation

Matching PR in forked repository

poorbarcode 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gaoran10 Jan 11, 2023 • edited Loading

Choose a reason for hiding this comment

coderzc Jan 11, 2023 • edited Loading

Choose a reason for hiding this comment

gaoran10 Jan 11, 2023 • edited Loading

Choose a reason for hiding this comment

mattisonchao Jan 13, 2023 • 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

coderzc Jan 16, 2023 • edited Loading

Choose a reason for hiding this comment

coderzc commented Jan 11, 2023 • edited Loading

gaoran10 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

Choose a reason for hiding this comment

mattisonchao left a comment • edited Loading

Choose a reason for hiding this comment

codecov-commenter commented Jan 13, 2023 • edited Loading

Codecov Report

gaoran10 Jan 11, 2023 •

edited

Loading

coderzc Jan 11, 2023 •

edited

Loading

gaoran10 Jan 11, 2023 •

edited

Loading

mattisonchao Jan 13, 2023 •

edited

Loading

coderzc Jan 16, 2023 •

edited

Loading

coderzc commented Jan 11, 2023 •

edited

Loading

mattisonchao left a comment •

edited

Loading

codecov-commenter commented Jan 13, 2023 •

edited

Loading