[Bug] version 2.8.0.7 Restarting broker consumes historical data

[sqtce]

Search before asking

• [X] I searched in the issues and found nothing similar.

Version

My brokers are using version 2.8.0.7 and there are three of them, used for handling delayed messages. Last night, due to a hardware issue, I restarted one of the brokers. As a result, all the messages that were supposed to be consumed by that broker on that day were re-delivered.

Minimal reproduce step

restarted

What did you expect to see?

How to resolve this issue?

What did you see instead?

re-delivered

Anything else?

No response

Are you willing to submit a PR?

• [X] I'm willing to submit a PR!

[JooHyukKim]

@sqtce Can you provide an appropriate title for easy understanding?

[sqtce]

Restarting broker consumes historical data

[tisonkun]

There is no 2.8.0.7 in upstream. Do you use the version of any vendors'?

Also, 2.8.x is end of life in upstream, the least maintained version is 2.10.4 now.

[github-actions[bot]]

The issue had no activity for 30 days, mark with Stale label.

[Ouriel]

Hello,

We are encountering the same behavior on our Pulsar cluster in 3.1. It's a topic with long retention time (90 days) with delayed message that are acknowledged once consumed. When restarting broker the pulsar backlog metric is getting back to the full backlog and we reprocess ~2M messages. Here is a grafana screenshot after the broker restart showing pulsar_subscription_back_log metrics on the topic

If you have a workaround or a solution that will be great!

[dao-jun]

Hello,

We are encountering the same behavior on our Pulsar cluster in 3.1. It's a topic with long retention time (90 days) with delayed message that are acknowledged once consumed. When restarting broker the pulsar backlog metric is getting back to the full backlog and we reprocess ~2M messages. Here is a grafana screenshot after the broker restart showing pulsar_subscription_back_log metrics on the topic

If you have a workaround or a solution that will be great!

@Ouriel What's your code looks like? I mean how do you create the consumers?

[Ouriel]

We are using pulsar through Infinitic.io framework Here is the consumer code : https://github.com/infiniticio/infinitic/blob/v0.13.0/infinitic-transport-pulsar/src/main/kotlin/io/infinitic/pulsar/consumers/Consumer.kt The setup of topic is also done by the framework here: https://github.com/infiniticio/infinitic/blob/v0.13.0/infinitic-transport-pulsar/src/main/kotlin/io/infinitic/pulsar/admin/PulsarInfiniticAdmin.kt