When is bad epoch message, need increase available permits.

apache · Oct 8, 2022 · 2d8c1e6 · 2d8c1e6
1 parent 63c1cbd
commit 2d8c1e6
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 1 deletion.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageRedeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageRedeliveryTest.java
@@ -313,6 +313,39 @@ public void testRedeliveryAddEpoch(boolean enableBatch) throws Exception{
         assertEquals(message.getValue(), test3);
     }
 
+    @Test(dataProvider = "enableBatch")
+    public void testRedeliveryAddEpochAndPermits(boolean enableBatch) throws Exception {
+        final String topic = "testRedeliveryAddEpochAndPermits";
+        final String subName = "my-sub";
+        // set receive queue size is 4, and first send 4 messages,
+        // then call redeliver messages, assert receive msg num.
+        int receiveQueueSize = 4;
+        ConsumerImpl<String> consumer = ((ConsumerImpl<String>) pulsarClient.newConsumer(Schema.STRING)
+                .topic(topic)
+                .receiverQueueSize(receiveQueueSize)
+                .autoScaledReceiverQueueSizeEnabled(false)
+                .subscriptionName(subName)
+                .subscriptionType(SubscriptionType.Failover)
+                .subscribe());
+
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+                .topic(topic)
+                .enableBatching(enableBatch)
+                .create();
+
+        consumer.setConsumerEpoch(1);
+        for (int i = 0; i < receiveQueueSize; i++) {
+            producer.send("pulsar" + i);
+        }
+        assertNull(consumer.receive(1, TimeUnit.SECONDS));
+
+        consumer.redeliverUnacknowledgedMessages();
+        for (int i = 0; i < receiveQueueSize; i++) {
+            Message<String> msg = consumer.receive();
+            assertEquals("pulsar" + i, msg.getValue());
+        }
+    }
+
     @Test(dataProvider = "enableBatch")
     public void testBatchReceiveRedeliveryAddEpoch(boolean enableBatch) throws Exception{
         final String topic = "testBatchReceiveRedeliveryAddEpoch";

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
@@ -462,7 +462,11 @@ protected Message<T> internalReceive() throws PulsarClientException {
     }
 
     boolean isValidConsumerEpoch(Message<T> message) {
-        return isValidConsumerEpoch((MessageImpl<T>) message);
+        if (!isValidConsumerEpoch((MessageImpl<T>) message)) {
+            increaseAvailablePermits(cnx());
+            return false;
+        }
+        return true;
     }
 
     @Override