[fix][broker] Fix silently dropped acknowledgement failures in PulsarMetadataEventSynchronizer - #26237

Motivation

PulsarMetadataEventSynchronizer.startConsumer()'s messageListener acknowledged consumed MetadataEvent messages with .thenApply(__ -> c.acknowledgeAsync(msg)). Since acknowledgeAsync itself returns a CompletableFuture, thenApply treated that return value as plain data instead of composing on it: the outer future completed as soon as the metadata event was applied to the local store, without ever waiting for the acknowledgement to actually finish.

As a result, if acknowledgeAsync later failed asynchronously (e.g. a connection issue while sending the ack command), the exception was never observed by the attached .exceptionally() handler and was silently dropped — no log
signal, with recovery relying entirely on the 60s ackTimeout redelivering the same event. The zero-listener branch had the same problem in an even more direct form: acknowledgeAsync was called with no error handling attached at all.

Modifications

• Extracted a shared acknowledgeAfter(CompletableFuture processed, Consumer c, Message msg) helper that uses thenCompose instead of thenApply, so the returned future genuinely waits for the acknowledgement and routes its failure into the existing .exceptionally() log handler.
• Applied the helper uniformly to all three code paths in the listener (zero, one, and multiple registered MetadataEvent listeners).

Verifying this change

• Make sure that the change passes the CI checks.
  This change is a trivial rework / code cleanup without any test coverage.
  (Note: the fix was manually verified locally with a temporary unit test that mocked Consumer.acknowledgeAsync and asserted the returned future doesn't settle until the ack future completes — confirmed it failed against the
  original thenApply shape and passed after switching to thenCompose. That test was not included in this PR.)
  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