(pulsar) branch master updated: [fix][test] Fix flaky V5ProducerBackpressureTest.retriesAcrossASegmentSplitKeepTheAccountingBalanced (#26792)

[lh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 86ba7ebf6a3b [fix][test] Fix flaky V5ProducerBackpressureTest.retriesAcrossASegmentSplitKeepTheAccountingBalanced (#26792)
86ba7ebf6a3b is described below

commit 86ba7ebf6a3be14e6410080ddcd9de18945e380f
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Oct 1 05:23:48 2026 +0300

    [fix][test] Fix flaky V5ProducerBackpressureTest.retriesAcrossASegmentSplitKeepTheAccountingBalanced (#26792)
---
 .../apache/pulsar/client/impl/v5/V5ProducerBackpressureTest.java | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerBackpressureTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerBackpressureTest.java
index b784eb9c2b5e..298e0bb09d73 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerBackpressureTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerBackpressureTest.java
@@ -387,9 +387,12 @@ public class V5ProducerBackpressureTest extends V5ClientBaseTest {
         List<CompletableFuture<MessageId>> futures = new ArrayList<>(numMessages);
         Thread sender = new Thread(() -> {
             for (int i = 0; i < numMessages; i++) {
-                CompletableFuture<MessageId> future = producer.async().newMessage().value(payload).send();
-                future.whenComplete((__, ___) -> completed.incrementAndGet());
-                futures.add(future);
+                // Keep the stage that counts the completion, so that waiting for the futures below
+                // also waits for the count. A future runs its dependents in LIFO order, so the
+                // allOf relay would otherwise run before the counter and could let the main thread
+                // read the count too early.
+                futures.add(producer.async().newMessage().value(payload).send()
+                        .whenComplete((__, ___) -> completed.incrementAndGet()));
             }
         }, "sender");
         sender.start();