Flaky-test: V5ConsumerPriorityLevelTest > testLowerPriorityConsumerReceivesOverflow

[lhotari]

Search before reporting

• I searched in the issues and found nothing similar.

Example failure

https://github.com/apache/pulsar/actions/runs/25426612333/job/74582655201?pr=25692#step:11:249

Exception stacktrace

Gradle suite > Gradle test > org.apache.pulsar.client.api.v5.V5ConsumerPriorityLevelTest > testLowerPriorityConsumerReceivesOverflow FAILED
      java.lang.AssertionError: low-priority consumer must see overflow once high consumer's prefetch fills, got: 0 expected [true] but found [false]
          at org.testng.Assert.fail(Assert.java:111)
          at org.testng.Assert.failNotEquals(Assert.java:1590)
          at org.testng.Assert.assertEqualsImpl(Assert.java:150)
          at org.testng.Assert.assertEquals(Assert.java:132)
          at org.testng.Assert.assertEquals(Assert.java:1079)
          at org.testng.Assert.assertEquals(Assert.java:1043)
          at org.apache.pulsar.client.api.v5.V5ConsumerPriorityLevelTest.testLowerPriorityConsumerReceivesOverflow(V5ConsumerPriorityLevelTest.java:158)

Are you willing to submit a PR?

• I'm willing to submit a PR!

[Dream95]

I think the flakiness is likely caused by ScalableQueueConsumer.startReceiveLoop().

ScalableQueueConsumer currently keeps calling v4Consumer.receiveAsync() in a recursive loop and pushes messages into a local LinkedTransferQueue (effectively unbounded).
This means even when the user code does not call receive() (or does not ack), the internal loop can still keep draining from the underlying v4 consumer and replenishing permits indirectly.

As a result:

1. the high-priority consumer may continue to accumulate messages beyond the expected user-level backpressure semantics;
2. overflow to the low-priority consumer becomes timing-dependent ,flaky
3. this may also create memory growth risk on the client side.

I think should add backpressure in ScalableQueueConsumer.