fix some test

congbobo184 · Nov 20, 2024 · ddc7def · ddc7def
1 parent e6f4e9a
commit ddc7def
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 4 deletions.
diff --git a/...org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/...org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -279,7 +279,7 @@ public void redeliverUnacknowledgedMessages(Consumer consumer, long consumerEpoc
     }
 
     private synchronized void internalRedeliverUnacknowledgedMessages(Consumer consumer, long consumerEpoch) {
-        // broker side epoch is smaller than consumer epoch, so don't need to handle this redeliver request
+        // broker side epoch is bigger than consumer epoch, so don't need to handle this redeliver request
         if (consumerEpoch < consumer.getConsumerEpoch()) {
             log.warn("[{}-{}] Ignoring redeliverUnacknowledgedMessages since broker epoch [{}] is smaller than "
                             + "consumer epoch [{}]",

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
@@ -345,12 +345,13 @@ public void testRedeliveryBrokerIgnoreSmallerEpoch() throws Exception{
                 "persistent://public/default/testRedeliveryBrokerAbortSmallerEpoch", false).get().get();
         Awaitility.await().until(() -> persistentTopic.getSubscription(subName).getDispatcher()
                 .getConsumers().get(0).getConsumerEpoch() == 1);
-        consumer.setConsumerEpoch(0);
+        consumer.setConsumerEpoch(-1);
         producer.send("Hello Pulsar!");
 
         // ignore this redeliver request
         consumer.redeliverUnacknowledgedMessages();
         consumer.receive();
+        assertEquals(consumer.getConsumerEpoch(), 0);
         assertEquals(persistentTopic.getSubscription(subName).getDispatcher()
                 .getConsumers().get(0).getConsumerEpoch(), 1);
     }
@@ -377,8 +378,7 @@ public void testRedeliveryCommandDontCheckClientConnectionState() throws Excepti
         producer.send("Hello Pulsar!");
         consumer.receive();
         consumer.redeliverUnacknowledgedMessages();
-        PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService().getTopic(
-                "persistent://public/default/testRedeliveryCommandDontCheckClientConnectionState",
+        PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService().getTopic(topic,
                 false).get().get();
         Awaitility.await().until(() -> persistentTopic.getSubscription(subName).getDispatcher()
                 .getConsumers().get(0).getConsumerEpoch() == 1);