[fix][broker] Correct multiple race conditions in PersistentDispatcherMultipleConsumers#25681
Draft
chamons wants to merge 1 commit intoapache:masterfrom
Draft
[fix][broker] Correct multiple race conditions in PersistentDispatcherMultipleConsumers#25681chamons wants to merge 1 commit intoapache:masterfrom
chamons wants to merge 1 commit intoapache:masterfrom
Conversation
…rMultipleConsumers - apache#25617 DelayedDeliveryTracker is not thread safe, and any access to it must be done holding the object lock. There were five cases I found, three of them I could correct just by adding synchronized to the method declaration. Two of them were overrides and were corrected using a manual locking scope. This is a significant issue, as exceptions here were uncaught and put the broker into an invalid state that prevented some delayed messages from being delivered until restart. The included unit test failed 100% of the time when run locally without the fix, and we were able to run https://github.com/chamons/pulsar-scheduled-exception-repro without issues with the fix included.
chamons
commented
May 5, 2026
| } | ||
|
|
||
| @Test | ||
| public void testRaceConditionInTrackDelayedDelivery() throws Exception { |
Author
There was a problem hiding this comment.
This is based on pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerThreadSafetyTest.java
merlimat
reviewed
May 5, 2026
| @Override | ||
| public long getNumberOfDelayedMessages() { | ||
| return delayedDeliveryTracker.map(DelayedDeliveryTracker::getNumberOfDelayedMessages).orElse(0L); | ||
| synchronized (this) { |
Contributor
There was a problem hiding this comment.
nit: this could be synchronized on the method itself
| .setPublishTime(System.currentTimeMillis()) | ||
| .setDeliverAtTime(deliverAt); | ||
|
|
||
| ExecutorService executorService = Executors.newFixedThreadPool(32); |
Contributor
There was a problem hiding this comment.
These are many threads for a unit test :)
The executor needs to get shutdown. Use @Cleanup("shutdown")
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
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.
Fixes #25617
Motivation
DelayedDeliveryTracker is not thread safe, and any access to it must be done holding the object lock.
This is a significant issue, as exceptions here were uncaught and put the broker into an invalid state that prevented some delayed messages from being delivered until restart.
Modifications
There were five cases I found, three of them I could correct just by adding synchronized to the method declaration. Two of them were overrides and were corrected using a manual locking scope.
Verifying this change
The included unit test failed 100% of the time when run locally without the fix, and we were able to run https://github.com/chamons/pulsar-scheduled-exception-repro without issues with the fix included.
Does this pull request potentially affect one of the following parts:
If the box was checked, please highlight the changes