This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 403ee7f23e53 [fix][client] Scalable topics: stream consumer must not
acknowledge on close or topic-removal (#26172)
403ee7f23e53 is described below
commit 403ee7f23e534611f570280fc547aff81ef99fd0
Author: Matteo Merli <[email protected]>
AuthorDate: Fri Jul 10 07:00:25 2026 -0700
[fix][client] Scalable topics: stream consumer must not acknowledge on
close or topic-removal (#26172)
---
.../api/v5/V5MultiTopicStreamConsumerTest.java | 75 ++++++++++++++++++++++
.../client/impl/v5/MultiTopicStreamConsumer.java | 29 +++++----
2 files changed, 90 insertions(+), 14 deletions(-)
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5MultiTopicStreamConsumerTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5MultiTopicStreamConsumerTest.java
index f3299b1c74be..16e89ebbc54c 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5MultiTopicStreamConsumerTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5MultiTopicStreamConsumerTest.java
@@ -29,6 +29,7 @@ import java.util.UUID;
import lombok.Cleanup;
import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.awaitility.Awaitility;
import org.testng.annotations.Test;
/**
@@ -176,6 +177,80 @@ public class V5MultiTopicStreamConsumerTest extends
V5ClientBaseTest {
+ (stale != null ? stale.value() : ""));
}
+ @Test
+ public void closeWithoutAckDoesNotAcknowledgeAnything() throws Exception {
+ // A stream consumer acknowledges only via explicit
acknowledgeCumulative. Closing
+ // the consumer (or, equivalently, a crash) must NOT acknowledge
anything on the
+ // application's behalf. Regression test for a close-time flush that
used to ack each
+ // per-topic consumer up to its prefetch frontier — silently acking
messages the
+ // application had buffered but never received. We assert it at the
broker: after
+ // receiving every message but acknowledging none, the subscription
backlog must
+ // still cover every message. (With the old flush this dropped to 0.)
+ String topicA = topicName("a");
+ String topicB = topicName("b");
+ admin.scalableTopics().createScalableTopic(topicA, 1);
+ admin.scalableTopics().createScalableTopic(topicB, 1);
+
+ @Cleanup
+ Producer<String> pa =
v5Client.newProducer(Schema.string()).topic(topicA).create();
+ @Cleanup
+ Producer<String> pb =
v5Client.newProducer(Schema.string()).topic(topicB).create();
+
+ String subscription = "multi-stream-close-no-ack";
+ StreamConsumer<String> first =
v5Client.newStreamConsumer(Schema.string())
+ .namespace(getNamespace())
+ .subscriptionName(subscription)
+
.subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+
+ int n = 5;
+ for (int i = 0; i < n; i++) {
+ pa.newMessage().value("a-" + i).send();
+ pb.newMessage().value("b-" + i).send();
+ }
+
+ // Receive every message — advancing each per-topic prefetch frontier
— but ack
+ // NOTHING. If close flushed that frontier, all 2n would be
acknowledged here.
+ Set<String> receivedFirst = new HashSet<>();
+ long deadline = System.currentTimeMillis() + 20_000L;
+ while (receivedFirst.size() < 2 * n && System.currentTimeMillis() <
deadline) {
+ Message<String> msg = first.receive(Duration.ofSeconds(1));
+ if (msg != null) {
+ receivedFirst.add(msg.value());
+ }
+ }
+ assertEquals(receivedFirst.size(), 2 * n, "first consumer should
receive every message");
+ first.close();
+
+ // At the broker, every message must still be in the backlog: close
acked nothing.
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(subscriptionBacklog(subscription, topicA,
topicB), 2L * n,
+ "close must not acknowledge anything; full backlog
must remain"));
+ }
+
+ /**
+ * Total delivered-but-unacked backlog for {@code subscription} across
every segment of
+ * the given scalable topics. Reads the broker's Topic reference directly
— the topics
+ * REST admin does not serve the {@code segment://} domain.
+ */
+ private long subscriptionBacklog(String subscription, String...
scalableTopics) throws Exception {
+ long total = 0;
+ for (String topic : scalableTopics) {
+ var stats = admin.scalableTopics().getStats(topic);
+ for (var seg : stats.getSegments().values()) {
+ var ref = getTopicReference(seg.name());
+ if (ref.isEmpty()) {
+ continue;
+ }
+ var sub = ref.get().getSubscription(subscription);
+ if (sub != null) {
+ total += sub.getNumberOfEntriesInBacklog(true);
+ }
+ }
+ }
+ return total;
+ }
+
@Test
public void filtersByPropertySoOnlyMatchingTopicsAttach() throws Exception
{
String aliceTopic = topicName("alice");
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
index a743d1882173..951de7980a19 100644
---
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
@@ -57,9 +57,10 @@ import org.apache.pulsar.common.naming.TopicName;
* per-topic consumer with the right segment vector — same semantics as the
* single-topic case, just lifted one level.
*
- * <p>For Removed-mid-stream topics we flush acks up to {@code latestDelivered}
- * for that topic before closing the per-topic consumer, so the user's
- * processing-acked invariant is preserved if the topic is later re-added.
+ * <p>Acknowledgment is always explicit: neither {@code close()} nor a topic
+ * leaving the matching set acks anything on the application's behalf. A
removed
+ * topic's per-topic consumer is simply detached; anything
delivered-but-unacked
+ * is redelivered if the topic is later re-added (at-least-once).
*/
final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
@@ -247,10 +248,13 @@ final class MultiTopicStreamConsumer<T> implements
StreamConsumer<T> {
}
/**
- * Close per-topic consumer, flushing pending cumulative acks up to
whatever was
- * last delivered for that topic. If the topic later re-appears
(re-Added), a
- * fresh consumer subscribes and resumes from the broker-side cursor —
already
- * advanced past the messages we've delivered to the user.
+ * Detach the per-topic consumer and drop our delivery tracking for it.
Runs both on
+ * {@link #closeAsync()} and when a topic leaves the matching set. We
deliberately do
+ * <em>not</em> acknowledge anything here: acks on a stream consumer are
cumulative and
+ * always explicit, so closing (or a topic removal) must never advance a
cursor past
+ * what the application itself acked. Whatever was delivered-but-unacked
is redelivered
+ * on the next attach (at-least-once). If the topic later re-appears, a
fresh consumer
+ * subscribes and resumes from the broker-side cursor.
*/
private CompletableFuture<Void> closeTopic(String topicName) {
retryDelays.remove(topicName);
@@ -264,12 +268,8 @@ final class MultiTopicStreamConsumer<T> implements
StreamConsumer<T> {
if (state == null) {
return CompletableFuture.completedFuture(null);
}
- // Flush: ack everything we delivered for this topic.
- Map<Long, org.apache.pulsar.client.api.MessageId> latest =
- latestDeliveredPerTopicSegment.remove(topicName);
- if (latest != null && !latest.isEmpty()) {
- state.consumer.ackUpToVector(latest);
- }
+ // Stop tracking this topic's delivery positions. No ack flush — see
javadoc.
+ latestDeliveredPerTopicSegment.remove(topicName);
return state.consumer.closeAsync()
.thenRun(() -> log.info().attr("topic", topicName)
.log("Per-topic stream consumer detached"));
@@ -337,7 +337,8 @@ final class MultiTopicStreamConsumer<T> implements
StreamConsumer<T> {
for (var entry : vector.entrySet()) {
PerTopic<T> state = perTopic.get(entry.getKey());
if (state == null) {
- // Topic was Removed since enqueue; closeTopic already flushed.
+ // Topic left the matching set since this message was
enqueued: we've
+ // detached it and no longer ack removed topics, so skip its
slice.
continue;
}
action.accept(state.consumer, entry.getValue());
