Flaky-test: PatternConsumerBackPressureTest.testInfiniteGetThousandsTopics

[lhotari]

Search before reporting

• I searched in the issues and found nothing similar.

Example failure

https://github.com/apache/pulsar/actions/runs/18322485242/job/52209927567?pr=24825#step:11:1728

Exception stacktrace

Error:  Tests run: 2, Failures: 1, Errors: 0, Skipped: 1, Time elapsed: 49.69 s <<< FAILURE! -- in org.apache.pulsar.client.api.PatternConsumerBackPressureTest
  Error:  org.apache.pulsar.client.api.PatternConsumerBackPressureTest.testInfiniteGetThousandsTopics -- Time elapsed: 15.35 s <<< FAILURE!
  org.awaitility.core.ConditionTimeoutException: Assertion condition expected [2048] but found [0] within 10 seconds.
  	at org.awaitility.core.ConditionAwaiter.await(ConditionAwaiter.java:167)
  	at org.awaitility.core.AssertionCondition.await(AssertionCondition.java:119)
  	at org.awaitility.core.AssertionCondition.await(AssertionCondition.java:31)
  	at org.awaitility.core.ConditionFactory.until(ConditionFactory.java:985)
  	at org.awaitility.core.ConditionFactory.untilAsserted(ConditionFactory.java:769)
  	at org.apache.pulsar.client.api.PatternConsumerBackPressureTest.testInfiniteGetThousandsTopics(PatternConsumerBackPressureTest.java:95)
  	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
  	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
  	at org.testng.internal.invokers.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:139)
  	at org.testng.internal.invokers.InvokeMethodRunnable.runOne(InvokeMethodRunnable.java:47)
  	at org.testng.internal.invokers.InvokeMethodRunnable.call(InvokeMethodRunnable.java:76)
  	at org.testng.internal.invokers.InvokeMethodRunnable.call(InvokeMethodRunnable.java:11)
  	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
  	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
  	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
  	at java.base/java.lang.Thread.run(Thread.java:1583)
  Caused by: java.lang.AssertionError: expected [2048] but found [0]
  	at org.testng.Assert.fail(Assert.java:110)
  	at org.testng.Assert.failNotEquals(Assert.java:1577)
  	at org.testng.Assert.assertEqualsImpl(Assert.java:149)
  	at org.testng.Assert.assertEquals(Assert.java:131)
  	at org.testng.Assert.assertEquals(Assert.java:1418)
  	at org.testng.Assert.assertEquals(Assert.java:1382)
  	at org.testng.Assert.assertEquals(Assert.java:1428)
  	at org.apache.pulsar.client.api.PatternConsumerBackPressureTest.lambda$testInfiniteGetThousandsTopics$2(PatternConsumerBackPressureTest.java:96)
  	at org.awaitility.core.AssertionCondition.lambda$new$0(AssertionCondition.java:53)
  	at org.awaitility.core.ConditionAwaiter$ConditionPoller.call(ConditionAwaiter.java:248)
  	at org.awaitility.core.ConditionAwaiter$ConditionPoller.call(ConditionAwaiter.java:235)
  	... 4 more

Are you willing to submit a PR?

• I'm willing to submit a PR!

[lhotari]

@poorbarcode This test was introduced in #24423, please check it.

[lhotari]

@poorbarcode Another failure here: https://github.com/apache/pulsar/actions/runs/18463783817/job/52601614708?pr=24840#step:11:1752

  Error:  Tests run: 2, Failures: 1, Errors: 0, Skipped: 1, Time elapsed: 97.05 s <<< FAILURE! -- in org.apache.pulsar.client.api.PatternConsumerBackPressureTest
  Error:  org.apache.pulsar.client.api.PatternConsumerBackPressureTest.testInfiniteGetThousandsTopics -- Time elapsed: 60.01 s <<< FAILURE!
  org.testng.internal.thread.ThreadTimeoutException: Method org.apache.pulsar.client.api.PatternConsumerBackPressureTest.testInfiniteGetThousandsTopics() didn't finish within the time-out 60000
  	at java.base/jdk.internal.misc.Unsafe.park(Native Method)
  	at java.base/java.util.concurrent.locks.LockSupport.park(LockSupport.java:221)
  	at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:754)
  	at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1099)
  	at java.base/java.util.concurrent.CountDownLatch.await(CountDownLatch.java:230)
  	at org.apache.pulsar.client.api.PatternConsumerBackPressureTest.testInfiniteGetThousandsTopics(PatternConsumerBackPressureTest.java:94)
  	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
  	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
  	at org.testng.internal.invokers.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:139)
  	at org.testng.internal.invokers.InvokeMethodRunnable.runOne(InvokeMethodRunnable.java:47)
  	at org.testng.internal.invokers.InvokeMethodRunnable.call(InvokeMethodRunnable.java:76)
  	at org.testng.internal.invokers.InvokeMethodRunnable.call(InvokeMethodRunnable.java:11)
  	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
  	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
  	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
  	at java.base/java.lang.Thread.run(Thread.java:1583)

the logs show io.netty.util.internal.OutOfDirectMemoryError

2025-10-13T11:36:02,114 - WARN  - [pulsar-client-io-6772-3:ClientCnx] - [localhost/127.0.0.1:35867] Got exception io.netty.util.internal.OutOfDirectMemoryError: failed to allocate 4194304 byte(s) of direct memory (used: 1358954503, max: 1363148800)
	at io.netty.util.internal.PlatformDependent.incrementMemoryCounter(PlatformDependent.java:880)
	at io.netty.util.internal.PlatformDependent.allocateDirectNoCleaner(PlatformDependent.java:809)
	at io.netty.buffer.PoolArena$DirectArena.allocateDirect(PoolArena.java:718)
	at io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:693)
	at io.netty.buffer.PoolArena.allocateNormal(PoolArena.java:213)
	at io.netty.buffer.PoolArena.tcacheAllocateNormal(PoolArena.java:195)
	at io.netty.buffer.PoolArena.allocate(PoolArena.java:137)
	at io.netty.buffer.PoolArena.reallocate(PoolArena.java:317)
	at io.netty.buffer.PooledByteBuf.capacity(PooledByteBuf.java:123)
	at io.netty.buffer.AbstractByteBuf.ensureWritable0(AbstractByteBuf.java:305)
	at io.netty.buffer.AbstractByteBuf.ensureWritable(AbstractByteBuf.java:280)
	at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1103)
	at io.netty.buffer.WrappedByteBuf.writeBytes(WrappedByteBuf.java:792)
	at io.netty.buffer.AdvancedLeakAwareByteBuf.writeBytes(AdvancedLeakAwareByteBuf.java:606)
	at io.netty.handler.codec.ByteToMessageDecoder$1.cumulate(ByteToMessageDecoder.java:105)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:288)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.handler.flush.FlushConsolidationHandler.channelRead(FlushConsolidationHandler.java:152)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1357)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)
	at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
	at io.netty.channel.epoll.AbstractEpollChannel$AbstractEpollUnsafe$1.run(AbstractEpollChannel.java:425)
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
	at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:405)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:1583)

It seems that the solution doesn't work?

logs are available at https://gist.github.com/lhotari/bb95ba67ebaa8b0adef1623c01746b9b

[lhotari]

The problem would be addressed by PIP-442 implementation PR, #24833

[poorbarcode]

It seems that the solution doesn't work?

The test will fail if it encounters the following 2 cases

• There is a bug that the client side may lose packages that the broker responded to when the traffic is really high. I am not sure whether it is a broker-side bug or a client-side bug.
• If the requests are sent at the same time, the broker will cache too many in-flight requests, which leads to the mechanism being unavailable.

The logs you shared is related to the case 1, I will work on it next week. The second situation should not occur in a real production environment. We can ignore it for now (just keep the test code stable), I will also work on it next week

[lhotari]

@poorbarcode There are lots of failures in CI. I think that we need to address this by disabling the test or by merging PR #24833.
Some links to failures here:
https://github.com/lhotari/pulsar-flakes/blob/master/2025-10-07-to-2025-10-14/org.apache.pulsar.client.api.PatternConsumerBackPressureTest.testInfiniteGetThousandsTopics.md
https://github.com/lhotari/pulsar-flakes/blob/master/2025-10-14-14d-master/org.apache.pulsar.client.api.PatternConsumerBackPressureTest.testInfiniteGetThousandsTopics.md

[BewareMyPower]

+1 to disabling this test. This test is very flaky

2025-11-05T20:11:35,097 - INFO  - [pulsar-client-io-40-3:PatternConsumerBackPressureTest] - latch-count: 1662, succeed: 387
2025-11-05T20:11:35,099 - WARN  - [pulsar-client-io-40-3:ClientCnx] - [localhost/127.0.0.1:54498] Got exception io.netty.util.internal.OutOfDirectMemoryError: failed to allocate 4194304 byte(s) of direct memory (used: 1358954503, max: 1363148800)
	at io.netty.util.internal.PlatformDependent.incrementMemoryCounter(PlatformDependent.java:880)

[lhotari]

+1 to disabling this test. This test is very flaky

@BewareMyPower I'll push a PR

[lhotari]

@BewareMyPower PR to disable the test: #24948, please review