fix: demand pauser clear queue inside drain loop to avoid concurrent read access

[ozangunalp]

The issue seen in #2019 checks : MultiDemandPausingTckTest>PublisherVerification.required_spec317_mustNotSignalOnErrorWhenPendingAboveLongMaxValue

NPE in spsc queue :

java.lang.NullPointerException: Cannot invoke "java.util.concurrent.atomic.AtomicReferenceArray.length()" because "buf" is null
	at org.jctools.queues.atomic.AtomicQueueUtil.length(AtomicQueueUtil.java:88)
	at org.jctools.queues.atomic.unpadded.BaseSpscLinkedAtomicUnpaddedArrayQueue.newBufferPoll(BaseSpscLinkedAtomicUnpaddedArrayQueue.java:339)
	at org.jctools.queues.atomic.unpadded.BaseSpscLinkedAtomicUnpaddedArrayQueue.poll(BaseSpscLinkedAtomicUnpaddedArrayQueue.java:287)
	at org.jctools.queues.atomic.unpadded.SpscUnboundedAtomicUnpaddedArrayQueue.poll(SpscUnboundedAtomicUnpaddedArrayQueue.java:32)
	at java.base/java.util.AbstractQueue.clear(AbstractQueue.java:147)
	at io.smallrye.mutiny.operators.multi.MultiDemandPausingOp$PausableProcessor.clearQueue(MultiDemandPausingOp.java:173)
	at io.smallrye.mutiny.operators.multi.MultiDemandPausingOp$PausableProcessor.cancel(MultiDemandPausingOp.java:224)
	at io.smallrye.mutiny.operators.multi.MultiOperatorProcessor.cancelUpstream(MultiOperatorProcessor.java:139)
	at io.smallrye.mutiny.operators.multi.MultiOperatorProcessor.cancel(MultiOperatorProcessor.java:127)
	at io.smallrye.mutiny.operators.multi.MultiOperatorProcessor.cancelUpstream(MultiOperatorProcessor.java:139)
	at io.smallrye.mutiny.operators.multi.MultiEmitOnOp$MultiEmitOnProcessor.cancel(MultiEmitOnOp.java:147)
	at org.reactivestreams.FlowAdapters$ReactiveToFlowSubscription.cancel(FlowAdapters.java:187)

I was only able to reproduce the issue locally once. But I think it comes because we are calling the clear queue outside of the drain loop.

@jponge I think the correct fix this. But I need some assistance validating that this is the correct implementation.

We do not have a similar issue in other operators because they do not let clearing the queue externally – which in itself may be a mistake...

[codecov]

Codecov Report

❌ Patch coverage is 90.00000% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 89.18%. Comparing base (e728220) to head (95dfed7).
⚠️ Report is 12 commits behind head on main.

Files with missing lines	Patch %	Lines
...e/mutiny/operators/multi/MultiDemandPausingOp.java	90.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #2021      +/-   ##
============================================
+ Coverage     89.01%   89.18%   +0.16%     
- Complexity     3063     3112      +49     
============================================
  Files           409      412       +3     
  Lines         13105    13274     +169     
  Branches       1654     1684      +30     
============================================
+ Hits          11666    11838     +172     
+ Misses          814      812       -2     
+ Partials        625      624       -1     

Files with missing lines	Coverage Δ
...e/mutiny/operators/multi/MultiDemandPausingOp.java	88.13% <90.00%> (ø)

... and 15 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[jponge]

This looks definitely the right approach (processing an external signal in a drain loop for correctness), but I wonder if there isn't another race condition in the making 😃

[jponge]

We're good to go, thanks @ozangunalp