[fix][client] PIP-84: Skip processing a message in the message listener if the consumer epoch is no longer valid (#25007)

lhotari · web-flow · commit 9deee37f0b42 · 2025-11-24T14:35:55.000+02:00
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
@@ -1210,6 +1210,17 @@ protected void callMessageListener(Message<T> msg) {
             // after enabled message listener.
             receivedConsumer.increaseAvailablePermits((MessageImpl<?>) (msg instanceof TopicMessageImpl
                                 ? ((TopicMessageImpl<T>) msg).getMessage() : msg));
+
+            MessageImpl<T> innerMessage = (MessageImpl<T>) (msg instanceof TopicMessageImpl
+                    ? ((TopicMessageImpl<T>) msg).getMessage() : msg);
+            if (!receivedConsumer.isValidConsumerEpoch(innerMessage)) {
+                if (log.isDebugEnabled()) {
+                    log.debug("[{}][{}] Skipping processing message since the consumer epoch is not valid. {}", topic,
+                            subscription, msg.getMessageId());
+                }
+                return;
+            }
+
             MessageId id;
             if (this instanceof ConsumerImpl) {
                 id = MessageIdAdvUtils.discardBatch(msg.getMessageId());
