You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2026/10/15 06:09:52 UTC
(pulsar) branch master updated: [fix][client] Return flow-control permit when client seek by messageID (#26726)
This is an automated email from the ASF dual-hosted git repository.
codelipenghui 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 e89b71b49533 [fix][client] Return flow-control permit when client seek by messageID (#26726)
e89b71b49533 is described below
commit e89b71b495337306372b50763d16de0336abd591
Author: Rahul <84...@users.noreply.github.com>
AuthorDate: Thu Oct 15 11:39:45 2026 +0530
[fix][client] Return flow-control permit when client seek by messageID (#26726)
Co-authored-by: Rahul Prasad <ra...@flipkart.com>
Co-authored-by: Claude Opus 4.8 <no...@anthropic.com>
---
.../broker/service/SubscriptionSeekTest.java | 191 +++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 8 +
2 files changed, 199 insertions(+)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionSeekTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionSeekTest.java
index b4d61c0751ef..433e14381a04 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionSeekTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionSeekTest.java
@@ -30,7 +30,10 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import java.io.IOException;
import java.lang.reflect.Field;
+import java.nio.file.Files;
+import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
@@ -59,6 +62,9 @@ import org.apache.commons.lang3.ArrayUtils;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.EncryptionKeyInfo;
import org.apache.pulsar.client.api.InjectedClientCnxClientBuilder;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
@@ -73,6 +79,7 @@ import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.BatchMessageIdImpl;
import org.apache.pulsar.client.impl.ClientBuilderImpl;
import org.apache.pulsar.client.impl.ClientCnx;
+import org.apache.pulsar.client.impl.ConsumerImpl;
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.client.impl.TopicMessageIdImpl;
import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
@@ -1185,4 +1192,188 @@ public class SubscriptionSeekTest extends BrokerTestBase {
assertTrue(e.getCause().getMessage().contains("Only support seek by messageId or timestamp"));
}
}
+
+ /**
+ * Seeking exclusively to a messageId re-dispatches that boundary message, which the consumer
+ * filters out instead of delivering. The dropped message's flow-control permit must still be
+ * returned; otherwise, with receiverQueueSize=1, the single leaked permit exhausts the budget and
+ * the consumer stalls right after the seek, never delivering the message after the seek target.
+ */
+ @Test
+ public void testSeekBoundaryDropDoesNotLeakPermit() throws Exception {
+ final String topicName = "persistent://prop/ns-abc/seekBoundaryPermitLeak";
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName)
+ .enableBatching(false).create();
+
+ List<MessageId> ids = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ ids.add(producer.send(("seek-msg-" + i).getBytes()));
+ }
+
+ @Cleanup
+ org.apache.pulsar.client.api.Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName("my-sub")
+ .receiverQueueSize(1) // tiny budget: a single leaked permit stalls the consumer
+ .subscribe();
+
+ // Seek exclusively to index 1. The boundary message (index 1) is filtered; the messages
+ // after it (index 2, 3, 4) must still be deliverable.
+ consumer.seek(ids.get(1));
+
+ Message<byte[]> msg = consumer.receive(10, TimeUnit.SECONDS);
+ assertNotNull(msg, "consumer stalled after seek: the boundary-message drop leaked its permit "
+ + "and the receiverQueueSize=1 budget was exhausted");
+ assertEquals(msg.getValue(), "seek-msg-2".getBytes());
+ consumer.acknowledge(msg);
+ }
+
+ /**
+ * The chunked variant of {@link #testSeekBoundaryDropDoesNotLeakPermit}: the boundary message
+ * dropped on seek is a chunked message. The ConsumerImpl drop block credits the non-last chunks
+ * at arrival and must repay the last chunk's permit when the assembled message is filtered;
+ * otherwise, with receiverQueueSize=1, the consumer stalls and the next chunked message is never
+ * delivered.
+ */
+ @Test
+ public void testSeekBoundaryDropDoesNotLeakPermitForChunkedMessage() throws Exception {
+ final String topicName = "persistent://prop/ns-abc/seekBoundaryPermitLeakChunked";
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName)
+ .enableBatching(false)
+ .enableChunking(true)
+ .chunkMaxMessageSize(100) // force multi-chunk from a modest payload
+ .create();
+
+ // Each message is ~3 chunks at chunkMaxMessageSize=100.
+ byte[] payload = new byte[250];
+ Arrays.fill(payload, (byte) 'x');
+ List<MessageId> ids = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ ids.add(producer.send(payload));
+ }
+
+ @Cleanup
+ org.apache.pulsar.client.api.Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName("my-sub")
+ .receiverQueueSize(1)
+ .subscribe();
+
+ // Seek exclusively to the first chunked message; it is filtered as the boundary. The chunked
+ // message after it must still be delivered.
+ consumer.seek(ids.get(0));
+
+ Message<byte[]> msg = consumer.receive(15, TimeUnit.SECONDS);
+ assertNotNull(msg, "consumer stalled after seek: dropping the chunked boundary message leaked "
+ + "a flow-control permit and the receiverQueueSize=1 budget was exhausted");
+ assertEquals(msg.getMessageId(), ids.get(1));
+ consumer.acknowledge(msg);
+ }
+
+ /**
+ * Seeking to a batch message id makes the broker redeliver that entry with only the not-yet-acked
+ * batch indexes, and it charges the consumer one permit per such index (not per message in the
+ * batch). If the consumer then drops the entry (here an undecryptable batch consumed with
+ * ConsumerCryptoFailureAction.CONSUME), it must refund exactly the permits that were charged.
+ */
+ @Test
+ public void testSeekBoundaryDropRefundsOnlyChargedPermitsForPartiallyAckedBatch() throws Exception {
+ final String topicName = "persistent://prop/ns-abc/seekBoundaryPermitPartialAckSet";
+ final String subName = "my-sub";
+ final int batchSize = 10;
+ final int seekBatchIndex = 7; // broker redelivers indexes 7..9 -> charges 3 of 10 permits
+ final int receiverQueueSize = 8; // refund flush threshold is 4: a full 10 refund would flush
+
+ class TestKeyReader implements CryptoKeyReader {
+ private EncryptionKeyInfo read(String prefix, String keyName) {
+ try {
+ EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
+ keyInfo.setKey(Files.readAllBytes(Paths.get("./src/test/resources/certificate/" + prefix
+ + keyName)));
+ keyInfo.setMetadata(new HashMap<>());
+ return keyInfo;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ @Override
+ public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) {
+ return read("public-key.", keyName);
+ }
+
+ @Override
+ public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> keyMeta) {
+ return read("private-key.", keyName);
+ }
+ }
+
+ // No key reader on the consumer + CONSUME: the encrypted batch is delivered undecryptable and
+ // is never expanded into individual messages.
+ @Cleanup
+ org.apache.pulsar.client.api.Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName(subName)
+ .cryptoFailureAction(ConsumerCryptoFailureAction.CONSUME)
+ .receiverQueueSize(receiverQueueSize)
+ .subscribe();
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName)
+ .addEncryptionKey("client-rsa.pem")
+ .cryptoKeyReader(new TestKeyReader())
+ .enableBatching(true)
+ .batchingMaxMessages(batchSize)
+ // long delay so that all messages end up in a single entry
+ .batchingMaxPublishDelay(10, TimeUnit.SECONDS)
+ .create();
+
+ List<CompletableFuture<MessageId>> futures = new ArrayList<>();
+ for (int i = 0; i < batchSize; i++) {
+ futures.add(producer.sendAsync(("batch-msg-" + i).getBytes()));
+ }
+ List<MessageId> ids = new ArrayList<>();
+ for (CompletableFuture<MessageId> future : futures) {
+ ids.add(future.get());
+ }
+ assertTrue(ids.get(seekBatchIndex) instanceof BatchMessageIdImpl);
+ BatchMessageIdImpl producedId = (BatchMessageIdImpl) ids.get(seekBatchIndex);
+ assertEquals(producedId.getBatchIndex(), seekBatchIndex);
+ // Ids returned by the producer do not carry the batch size, but the seek's ackSet is built from
+ // it, so supply it explicitly.
+ BatchMessageIdImpl seekId = new BatchMessageIdImpl(producedId.getLedgerId(), producedId.getEntryId(),
+ producedId.getPartitionIndex(), seekBatchIndex, batchSize, null);
+
+ // Consume the initial dispatch of the undecryptable batch.
+ assertNotNull(consumer.receive(10, TimeUnit.SECONDS));
+
+ // Exclusive seek into the batch: the broker redelivers the entry with ackSet = indexes 7..9
+ // and the client drops it as the seek boundary.
+ consumer.seek(seekId);
+ Awaitility.await().until(consumer::isConnected);
+
+ PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topicName).get();
+ PersistentSubscription sub = topicRef.getSubscription(subName);
+ // The (reconnected) broker-side consumer has dispatched the redelivered entry.
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertFalse(sub.getConsumers().isEmpty());
+ assertTrue(sub.getConsumers().get(0).getMsgOutCounter() > 0);
+ });
+
+ // Nothing is queued for the application, so every permit must be accounted for: those the broker
+ // still holds plus those the client has refunded but not yet flushed. A missing refund leaves the
+ // total below receiverQueueSize (budget leaked); an over-refund pushes it above (over-credit).
+ ConsumerImpl<byte[]> consumerImpl = (ConsumerImpl<byte[]>) consumer;
+ Awaitility.await().during(2, TimeUnit.SECONDS).atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ int brokerPermits = sub.getConsumers().get(0).getAvailablePermits();
+ int clientUnflushed = consumerImpl.getAvailablePermits();
+ assertEquals(brokerPermits + clientUnflushed, receiverQueueSize,
+ "permit accounting is off after dropping a partially-acked batch (broker permits "
+ + brokerPermits + ", client unflushed " + clientUnflushed + ")");
+ });
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index c3dd27a47f3e..8a52295d48f1 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -1556,6 +1556,14 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
.log("Ignoring message from before the startMessageId");
uncompressedPayload.release();
+ // This message is dropped instead of being delivered to the application, so its
+ // outstanding flow-control permit is never returned via messageProcessed(). Refund
+ // exactly what the broker charged for this entry: one
+ // permit per message, minus the batch indexes it reports as already acked via ackSet (set
+ // bits are the still-unacked indexes). An undecryptable batch can also reach this block;
+ // for the plain and chunked cases there is no ackSet and numMessages is 1.
+ final int chargedPermits = ackSet.length > 0 ? BitSet.valueOf(ackSet).cardinality() : numMessages;
+ increaseAvailablePermits(cnx, chargedPermits);
return;
}