[improve][broker][PIP-195] Make bucket merge operation asynchronous #19873
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
coderzc
added
area/broker
type/enhancement
coderzc
force-pushed
the
improve/bucket_delayed_async_merge
branch
4 times, most recently
from
b2f9f49 to
3cc918d
Compare
coderzc
force-pushed
the
improve/bucket_delayed_async_merge
branch
from
3cc918d to
f854c59
Compare
poorbarcode
reviewed
Mar 23, 2023
| private volatile List<DelayedMessageIndexBucketSnapshotFormat.SnapshotSegment> snapshotSegments; | ||
| private List<DelayedMessageIndexBucketSnapshotFormat.SnapshotSegment> snapshotSegments; | ||
|
|
||
| boolean merging = false; |
There was a problem hiding this comment.
the variable merging should be modified with volatile.
There was a problem hiding this comment.
Because all the read and write of merging is in synchronized block.
There was a problem hiding this comment.
I think you are right, the document of J2SE describes this feature like this:
Intrinsic Locks and Synchronization
Synchronization is built around an internal entity known as the intrinsic lock or monitor lock. (The API specification often refers to this entity simply as a "monitor.") Intrinsic locks play a role in both aspects of synchronization: enforcing exclusive access to an object's state and establishing happens-before relationships that are essential to visibility.
Synchronized Methods
Second, when a synchronized method exits, it automatically establishes a happens-before relationship with any subsequent invocation of a synchronized method for the same object. This guarantees that changes to the state of the object are visible to all threads.
| @@ -341,18 +339,24 @@ public synchronized boolean addMessage(long ledgerId, long entryId, long deliver | |||
| private synchronized CompletableFuture<Void> asyncMergeBucketSnapshot() { | |||
| List<ImmutableBucket> values = immutableBuckets.asMapOfRanges().values().stream().toList(); | |||
| long minNumberMessages = Long.MAX_VALUE; | |||
| long minScheduleTimestampSum = Long.MAX_VALUE; | |||
There was a problem hiding this comment.
Is variable minScheduleTimestampSum unnecessary?
There was a problem hiding this comment.
Ah, it helps to find the first segment, right?
There was a problem hiding this comment.
Oh, it is used to find the bucket which the min scheduled time of the next segment to merge, avoid preload indexes that have not expired for a long time.
poorbarcode
reviewed
Mar 23, 2023
...oker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
Show resolved
Hide resolved
coderzc
force-pushed
the
improve/bucket_delayed_async_merge
branch
from
541ec76 to
133887f
Compare
poorbarcode
approved these changes
Mar 23, 2023
mattisonchao
approved these changes
Mar 24, 2023
coderzc
force-pushed
the
improve/bucket_delayed_async_merge
branch
from
8f5006b to
22c6b41
Compare
coderzc
force-pushed
the
improve/bucket_delayed_async_merge
branch
from
22c6b41 to
e2260cc
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
PIP: #16763
Motivation & Modifications
Verifying this change
(Please pick either of the following options)
This change is a trivial rework / code cleanup without any test coverage.
(or)
This change is already covered by existing tests, such as (please describe tests).
(or)
This change added tests and can be verified as follows:
(example:)
Documentation
docdoc-requireddoc-not-neededdoc-completeMatching PR in forked repository
PR in forked repository: