merlimat opened a new pull request, #26162:
URL: https://github.com/apache/pulsar/pull/26162
### Motivation
`MultiTopicStreamConsumer.onPerTopicMessage` built the per-message
cumulative-ack
position vector by deep-copying the entire cross-topic map on **every**
delivered
message, inside a global `synchronized` block:
```java
synchronized (latestDeliveredPerTopicSegment) {
snapshot = new HashMap<>(latestDeliveredPerTopicSegment.size());
for (var e : latestDeliveredPerTopicSegment.entrySet()) {
snapshot.put(e.getKey(), new HashMap<>(e.getValue())); // O(topics
× segments)
}
}
```
That is O(topics × segments) allocation and copy per message, and it
serialises
every per-topic netty IO thread through one monitor — directly undercutting
the
"scalable" goal of these consumers.
### Modifications
The message's own `positionVector` is already an **immutable** (`Map.copyOf`)
snapshot of every live segment of its topic up to that message. So:
- Adopt it directly as the topic's slice of the cross-topic map (no
per-segment
copy) and take a **shallow** copy of the cross-topic map, sharing those
immutable slices. Cost drops from O(topics × segments) to O(topics).
- Drop the lock: because each slice is immutable, a concurrent delivery to
another topic can only atomically swap in a whole new slice (`CHM.put`),
never
mutate one in place, so the shallow snapshot can never observe a torn
per-topic
view. The acking topic's own slice is exact (its `put` happens-before the
snapshot read on the same thread); other topics may be one version stale,
which
is a safe under-ack.
Replacing the per-topic slice (vs the old accumulate-forever `putAll`) drops
segments that have sealed or rebalanced away, but those are already absent
from
`segmentConsumers`, so `ScalableStreamConsumer.acknowledgeCumulative` already
skips them — the ack was a no-op either way, so behavior is unchanged.
### Verifying this change
Behavior-preserving refactor. Verified empirically: a mid-stream
cumulative-ack
scenario produces byte-for-byte identical outcomes on this change and on the
prior deep-copy implementation (same snapshot content). Existing coverage
stands:
`V5MultiTopicStreamConsumerTest.cumulativeAckCoversEveryTopicSeenSoFar`
(cross-topic cumulative ack), `V5CumulativeAckTest` (multi-segment
cumulative ack
incl. sealed-parent), and `MessageIdV5Test` (multiTopicVector serialization).
Full `pulsar-client-v5` suite passes (96 tests); checkstyle clean.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
