[fix][client] Scalable topics: stream consumer must not acknowledge on close or topic-removal

[merlimat]

Motivation

The namespace (multi-topic) v5 stream consumer violates at-least-once on close.

MultiTopicStreamConsumer.closeTopic() flushes a cumulative ack up to latestDeliveredPerTopicSegment for each topic before closing its per-topic consumer. That map tracks the latest message pumped into the client-side multiplexed queue — the prefetch frontier — not what the application actually received via receive(). It is the only site that reads the live map instead of a per-message frozen snapshot.

So closing the consumer — or any topic leaving the matching set — acknowledges every message prefetched into the mux, including messages the application never received. On a crash or clean close right after, those messages are silently lost. Measured with two 1-segment topics and 12 messages produced: receiving only one message, then closing (with acknowledgeCumulative never called), silently acks the other 11 — they are neither received by the consumer nor redelivered to a re-subscribe.

The single-topic ScalableStreamConsumer has no such flush and is unaffected. The explicit acknowledgeCumulative position vector is a frozen enqueue-time snapshot and, because the multiplexed queue is FIFO, is exact — it never covers unreceived messages. Only the close/removal flush is wrong.

Modifications

• Drop the ack flush from MultiTopicStreamConsumer.closeTopic(). Acknowledgment on a stream consumer is cumulative and always explicit: close() and a topic leaving the matching set now detach the per-topic consumer without acknowledging. Anything delivered-but-unacked is redelivered on the next attach (at-least-once). The per-topic delivery-tracking entry is still removed (memory hygiene).
• Clarify the related comment where a cumulative ack whose vector references a since-removed topic skips that topic's slice (behavior unchanged).

Verifying this change

This change added tests and can be verified as follows:

• V5MultiTopicStreamConsumerTest.closeWithoutAckDoesNotAcknowledgeAnything: produce to two topics, receive every message but acknowledge nothing, close, then assert at the broker that the full backlog remains — i.e. nothing was acknowledged on close. Verified to fail before this change (backlog drops to 0) and pass after.

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