fix

apache · Nov 15, 2022 · 07cd25c · 07cd25c
1 parent d237f50
commit 07cd25c
Show file tree

Hide file tree

Showing 3 changed files with 2 additions and 15 deletions.
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
@@ -429,12 +429,14 @@ public void testMultiConsumerRedeliveryAddEpoch(boolean enableBatch) throws Exce
         final String subName = "my-sub";
         admin.topics().createPartitionedTopic(topic, 5);
         final int messageNumber = 50;
+        @Cleanup
         Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
                 .topic(topic)
                 .subscriptionName(subName)
                 .subscriptionType(SubscriptionType.Failover)
                 .subscribe();
 
+        @Cleanup
         Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
                 .topic(topic)
                 .enableBatching(enableBatch)

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
@@ -945,10 +945,6 @@ protected final void notifyPendingBatchReceivedCallBack(CompletableFuture<Messag
             Message<T> msg = incomingMessages.poll();
             if (msg != null) {
                 messageProcessed(msg);
-                if (!isValidConsumerEpoch((MessageImpl<T>) msg)) {
-                    msgPeeked = incomingMessages.peek();
-                    continue;
-                }
                 Message<T> interceptMsg = beforeConsume(msg);
                 messages.add(interceptMsg);
             }

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
@@ -460,10 +460,6 @@ protected Message<T> internalReceive() throws PulsarClientException {
         }
     }
 
-    boolean isValidConsumerEpoch(Message<T> message) {
-        return isValidConsumerEpoch((MessageImpl<T>) message);
-    }
-
     @Override
     protected CompletableFuture<Message<T>> internalReceiveAsync() {
         CompletableFutureCancellationHandler cancellationHandler = new CompletableFutureCancellationHandler();
@@ -1921,13 +1917,6 @@ public void redeliverUnacknowledgedMessages() {
         }
     }
 
-    public int clearIncomingMessagesAndGetMessageNumber() {
-        int messagesNumber = incomingMessages.size();
-        clearIncomingMessages();
-        unAckedMessageTracker.clear();
-        return messagesNumber;
-    }
-
     @Override
     public void redeliverUnacknowledgedMessages(Set<MessageId> messageIds) {
         if (messageIds.isEmpty()) {