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

Author: merlimat

pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5MultiTopicStreamConsumerTest.java

@@ -29,6 +29,7 @@
 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 void cumulativeAckCoversEveryTopicSeenSoFar() throws Exception {
                         + (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");

pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java

@@ -57,9 +57,10 @@
  * 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 @@ private ConsumerConfigurationData<T> perTopicConf(String topicName) {
     }
 
     /**
-     * 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 @@ private CompletableFuture<Void> closeTopic(String topicName) {
         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 @@ private void fanOutCumulativeAck(MessageId messageId,
         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());