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

[lhotari]

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.