[Bug] MultiTopicsConsumer still received message with older ConsumerEpoch after redeliverUnacknowledgeMessages()

[YinY1]

Search before reporting

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

User environment

• broker version: 4.0.8
• client version: 4.1.1 (java)

Issue Description

configs

• producer config
  • enableBatching(true)
  • blockIfQueueFull(true)
  • .batchingMaxMessages(100)
• consumer config
  • subscriptionType(SubscriptionType.Exclusive)
  • subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
  • enableBatchIndexAcknowledgment(false)

What I did

• produce messages with property "sequence", then consume them and validate sequence number.
• cumulative acknowledge every groupSize message received
• redeliverUnacknowledgedMessages in schedule, and reset sequence number.

Expected

assume current epoch = 0
if lastAckNumber = 1000，and number of current received message is 1300 which matches expectedSequenceNumber, after calling redeliverUnacknowledgedMessages(), it should received a message with number 1001 with higher comsumerEpoch and redeliveryCount

Got

A random message with number > 1001, consumerEpoch = 0, redeliveryCount = 0
And other message with bigger batch index in the same batch(entry) filltered out.
see logs below.

Error messages

2026-02-03 02:58:43.202 [pool-2-thread-2] INFO  o.e.OrderedCumulativeConsumerWorker - After interval redeliver, resetting expected sequence to 4501
2026-02-03 02:58:43.202 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:76], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:77], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:78], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:79], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:80], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:81], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:82], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:83], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:84], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:85], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:86], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:87], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:88], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:89], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:90], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:91], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.205 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:92], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.206 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:93], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.206 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:94], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.206 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:95], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.206 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:96], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.206 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:97], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.206 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:98], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.206 [pulsar-client-internal-13-1] INFO  o.a.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://public/default/order], messageId : [14248:2059:-1:99], messageConsumerEpoch : [0], consumerEpoch : [1]
2026-02-03 02:58:43.703 [pool-2-thread-2] WARN  o.e.OrderedCumulativeConsumerWorker - Found out-of-order message: MessageId=14248:2059:-1:75, expectedSequence=4501, actualSequence=5976, uuid=d2363f75-f1db-453b-9b0c-e8d70b91ba0f, consumerEpoch=0, redeliveryCount=0

Reproducing the issue

  @SneakyThrows
  private void produceMessages(Producer<byte[]> producer, int count) {
      for (int i = 0; i < count; i++) {
          producer.newMessage().property("sequence", String.valueOf(i))
                  .value(("message-" + i).getBytes())
                  .send();
      }
  }

  @SneakyThrows
  private void consumeAndRedeliverRoundtrip(Consumer<byte[]> consumer, int groupSize, long interval) {
      long expectedSequenceNumber = 0;
      long lastAckNumer = 0;
      int count = 0;
      long last = System.currentTimeMillis();
      while (true) {
          var msg = consumer.receive(10, TimeUnit.SECONDS);
          if (msg == null) {
              log.info("No more messages to consume.");
              return;
          }

          // Validate sequence - should be strictly increasing, gap indicates message loss
          String sequenceStr = msg.getProperty("sequence");
          if (sequenceStr != null) {
              long actualSequence = Long.parseLong(sequenceStr);
              if (actualSequence != expectedSequenceNumber) {
                  log.error("Sequence gap detected! Expected: {}, Actual: {}", expectedSequenceNumber, actualSequence);
                  throw new RuntimeException("Sequence gap detected! Expected: " + expectedSequenceNumber
                      + ", Actual: " + actualSequence);
              }
          }
          count ++;
          if (count == groupSize) {
              consumer.acknowledgeCumulative(msg);
              count = 0;
              lastAckNumer = expectedSequenceNumber;
          }
          expectedSequenceNumber++;

          long cur = System.currentTimeMillis();
          if ( cur - last >= interval ) {
              consumer.redeliverUnacknowledgedMessages();
              expectedSequenceNumber = lastAckNumber + 1; // reset expected seq
              last = cur;
              Thread.sleep(100);
          }
      }
  }

Additional information

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[YinY1]

I suppose that this bug may caused by MultiTopicsConsumerImpl.receiveMessageFromConsumer().
More specificly, consumer async received a batch of messages, then validated consumer epoch. If redeliverUnacknowledgedMessages increses epoch concurrently, messages with same epoch may be filtered out after epoch updated, but some will not.

For example, a list of messages with epoch [0, 0, 0]. User call redeliverUnacknowledgedMessages updated epoch to 1, so user wants get messages with epoch 1 after called that.
But in internal consumer, messagesFuture may completed before user's call, then iterate each message and validate epoch. msg[0] will not be filtered out for current epoch is 0. Then user's call happened, epoch has been updated to 1, so msg[1..] will be filtered out.

timeline is like:
CONSUMER_EPOCH = 0 -->
messagesFuture.thenAcceptAsync with messages epochs [0, 0, 0] -->
isValidConsumerEpoch(messages[0]) = true -->
redeliverUnacknowledgedMessages() -->
CONSUMER_EPOCH.incrementAndGet (now is 1)
isValidConsumerEpoch(messages[1]) = false -->
isValidConsumerEpoch(messages[2]) = false -->
consumer.receive() got messages[0]

relevant code is in pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:250-284

[ChimdumebiNebolisa]

I’d like to take this issue. I’m going to reproduce it on a supported version, add a regression test, and then propose a fix. Please assign it to me, and I’ll make sure to post updates as I narrow down the root cause.

[ChimdumebiNebolisa]

Fixed by PR #25210.