[Bug] BucketDelayedDeliveryTracker.containsMessage is not thread-safe, but it's called from another thread

[lhotari]

Search before asking

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

Version

any released version of Pulsar

Minimal reproduce step

There's code that calls BucketDelayedDeliveryTracker.containsMessage.containsMessage from another thread

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

Lines 389 to 392 in 1c53841

	if (deliveryTracker instanceof BucketDelayedDeliveryTracker) {
	skipCondition = position -> ((BucketDelayedDeliveryTracker) deliveryTracker)
	.containsMessage(position.getLedgerId(), position.getEntryId());
	}

What did you expect to see?

Having thread-safety aspects considered in the BucketDelayedDeliveryTracker implementation.

What did you see instead?

There seem to be thread safety issues in BucketDelayedDeliveryTracker

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[lhotari]

Regarding thread safety, this tool mentioned in this blog post could be useful in finding concurrency bugs: https://www.elastic.co/search-labs/blog/optimistic-concurrency-lucene-debugging . The tool is Fray, https://github.com/cmu-pasta/fray .

[lhotari]

I wonder if #22716 is caused by the thread safety issues.

[lhotari]

@dao-jun @congbobo184 I'm reopening this issue since there are multiple remaining thread safety issues in BucketDelayedDeliveryTracker. For example nextDeliveryTime method is not thread safe. There are also other stats related methods which aren't thread safe.

Simply making all methods synchronized could result in deadlocks so that's why I think there should be concurrency tests in place to catch possible regressions that would be caused by adding synchronization. The comment #23190 (comment) references a tool that can be used to catch concurrency issues. However, that would also require useful assertions to check that results are correct under data races that the tool injects.

Instead of synchronization, it could be useful to use read-write locks if a synchronized block introduces a performance regression. In Pulsar, we have a JMH module where microbenchmarks could be written using JMH, it's in https://github.com/apache/pulsar/tree/master/microbench . Generative AI coding assistants are great in guiding in writing JMH test cases.

read-write locks could also cause deadlocks. One common issue with java.util.concurrent.locks.ReentrantReadWriteLock is that an acquired read-lock cannot be "upgraded" to a write lock later. java.util.concurrent.locks.StampedLock would be the most performant solution, but it's not re-entrant and that could also cause deadlocks.

[lhotari]

Re-opening this issue since #24542 didn't fix this yet.