Skip to content

[fix][client] Make V5 async producer flush() await the caller's send futures - #26287

Open
lhotari wants to merge 1 commit into
apache:masterfrom
lhotari:lh-fix-v5-flush-send-tracking
Open

[fix][client] Make V5 async producer flush() await the caller's send futures#26287
lhotari wants to merge 1 commit into
apache:masterfrom
lhotari:lh-fix-v5-flush-send-tracking

Conversation

@lhotari

@lhotari lhotari commented Aug 7, 2026

Copy link
Copy Markdown
Member

Fixes #26283

Motivation

V5AsyncApisTest.testAsyncProducerSendAndFlush fails intermittently in CI:

java.lang.AssertionError: send future 19 must be done after flush() expected [true] but found [false]
	at org.apache.pulsar.client.api.v5.V5AsyncApisTest.testAsyncProducerSendAndFlush(V5AsyncApisTest.java:83)

This is not test flakiness — it is a real contract violation in the V5 client.

AsyncMessageBuilderV5.send() returned

producer.sendInternalAsync(...).thenApply(id -> id);

an identity stage whose only purpose was to widen CompletableFuture<MessageIdV5> to
CompletableFuture<MessageId> (Java generics are invariant). Meanwhile
ScalableTopicProducer registered the upstream future in inFlightSends, and
flushAsync() returns CompletableFuture.allOf(...) over a snapshot of that set.

So allOf and the caller-facing thenApply stage were sibling dependents of the same
upstream future
. When the last send is acked on the Netty IO thread,
CompletableFuture.postComplete() drains the dependent stack in LIFO order: the allOf node
fires first, completing flush() and unparking the caller blocked in get(), before the
thenApply link that completes the future the caller actually holds. The caller can then
observe isDone() == false on a send that flush() claimed to have awaited.

The reported index is always the last one because allOf completes precisely on the last
send's completion, so that send's caller-facing stage is the one still pending. On an
unloaded machine the Netty thread wins that race by microseconds, which is why this only
showed up under CI load.

Modifications

  • ScalableTopicProducer.sendInternalAsync now creates and returns a
    CompletableFuture<MessageId> directly, and AsyncMessageBuilderV5.send() hands it back
    as-is. inFlightSends therefore holds the very futures returned to callers, so when
    flush()'s allOf completes, every one of them is guaranteed isDone()allOf only
    completes after each component's result is set. This also drops one allocation and one
    dependent stage per async send.
  • dispatchSendAttempt / handleAsyncSegmentFailure / inFlightSends follow the same type
    change. MessageIdV5 still implements org.apache.pulsar.client.api.v5.MessageId, so the
    segment-carrying message id is unchanged on the wire and in the completed value; only the
    static type of the internal plumbing narrows. sendInternalAsync had exactly one caller.
  • Comments on inFlightSends and send() record why no wrapper stage may be introduced
    between the tracked future and the returned one, so the bug is not reintroduced.
  • V5AsyncApisTest.testAsyncProducerSendAndFlush now samples isDone() from inside a
    flush() dependent instead of after flush().get() returns. That dependent runs on the
    thread completing flush(), before it moves on to any other dependent of the last send, so
    a send that flush() did not genuinely await shows up as not-done every time instead of
    only when the woken test thread happens to win the race. The original assertions are kept
    and none are weakened.

Verifying this change

  • Make sure that the change passes the CI checks.

This change is already covered by existing tests, such as
V5AsyncApisTest.testAsyncProducerSendAndFlush, which was strengthened here so that it
pins the regression deterministically.

Verified locally:

  • With the .thenApply(id -> id) wrapper temporarily reintroduced, the updated test fails
    deterministically with exactly the CI assertion
    (send future 19 must be done when flush() completes expected [true] but found [false])
    — i.e. the test now catches the bug on every run rather than occasionally.
  • With the fix in place, the whole V5AsyncApisTest class is green, with
    testAsyncProducerSendAndFlush at invocationCount = 10: 15/15 tests passed, 0 failures.
    (invocationCount was removed again before opening this PR.)
  • The ordering hazard was also confirmed in isolation with a standalone JDK harness modelling
    both shapes over 20 futures × 2000 runs: the old shape (identity thenApply + allOf on the
    upstream future) had at least one caller-visible future not done at flush completion in
    2000/2000 runs; the new shape, 0/2000.
  • ./gradlew spotlessCheck checkstyleMain checkstyleTest passes.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

  • Dependencies (add or upgrade a dependency)
  • The public API
  • The schema
  • The default values of configurations
  • The threading model
  • The binary protocol
  • The REST endpoints
  • The admin CLI options
  • The metrics
  • Anything that affects deployment

The V5 public API is unchanged: AsyncMessageBuilder.send() keeps its
CompletableFuture<MessageId> signature, and the completed value is still a MessageIdV5.
Only which CompletableFuture instance is returned changes, which is what fixes the
flush() guarantee.

…futures

ScalableTopicProducer tracked the upstream future in inFlightSends while
AsyncMessageBuilderV5.send() handed callers a derived thenApply(id -> id)
stage that existed only to widen MessageIdV5 to MessageId. flushAsync()'s
allOf and the caller-facing stage were therefore sibling dependents of the
same upstream future, and the JDK fires dependents in unspecified order, so
flush() could complete while a send future the caller holds was still not
isDone().

Return the producer's future directly instead, typed as
CompletableFuture<MessageId>, so flush() awaits exactly the futures callers
observe. Also sample isDone() from inside a flush() dependent in the test so
the contract is checked deterministically rather than by timing luck.

Fixes apache#26283

Assisted-by: Claude Code (Opus 5)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Flaky-test: V5AsyncApisTest.testAsyncProducerSendAndFlush

1 participant