You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/12 19:18:44 UTC
(pulsar) 19/27: [fix][test] Fix flaky V5ProducerBackpressureTest.asyncSendsBlockTheCallerAtTheMemoryLimit (#26855)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ce7b8011f885f3b702a167e46d33e2274f81bb08
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Mon Oct 12 08:57:33 2026 -0700
[fix][test] Fix flaky V5ProducerBackpressureTest.asyncSendsBlockTheCallerAtTheMemoryLimit (#26855)
(cherry picked from commit 56765568977a071f4a13c43ae8512d8f8728b552)
---
.../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 298e0bb09d73..8fa9f01d3bd7 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
@@ -122,9 +122,12 @@ public class V5ProducerBackpressureTest extends V5ClientBaseTest {
List<CompletableFuture<MessageId>> futures = new ArrayList<>(NUM_MESSAGES);
int maxOutstanding = 0;
for (int i = 0; i < NUM_MESSAGES; i++) {
- CompletableFuture<MessageId> future = 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(async.newMessage().value(payload).send()
+ .whenComplete((__, ___) -> completed.incrementAndGet()));
// The first sends go out before the segment producer even exists. Without admission on
// the caller's thread they would all queue up here, unbounded.
maxOutstanding = Math.max(maxOutstanding, i + 1 - completed.get());