[Bug] Infinite message loop after restart

### Search before asking

- [X] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.

### Version

Client - 3.1.0
Pulsar - 3.1.0 (and later builds)

### Minimal reproduce step

Publishing 1 000 000 messages. Each message 30Kb. Servers restarts in the meantime.

### What did you expect to see?

All messages read and backlog is empty

### What did you see instead?

On one of the partitions the backlog has messages. Consumer keeps reading (indefinitely) the messages and sends ack successfully. Backlog remains the same

### Anything else?

After restart there is a rewind [](url)

Partition stats show readPosition on different ledger

kubectl exec --namespace pulsar pulsar-toolset-0 -- bin/pulsar-admin topics stats persistent://websight/dxp/search_experiences-partition-3
{
“msgRateIn” : 0.0,
“msgThroughputIn” : 0.0,
“msgRateOut” : 1009.3990130769382,
“msgThroughputOut” : 1.5503941574612858E7,
“bytesInCounter” : 0,
“msgInCounter” : 0,
“bytesOutCounter” : 4276153214,
“msgOutCounter” : 278397,
“averageMsgSize” : 0.0,
“msgChunkPublished” : false,
“storageSize” : 4842312348,
“backlogSize” : 4842296220,
“publishRateLimitedTimes” : 0,
“earliestMsgPublishTimeInBacklogs” : 0,
“offloadedStorageSize” : 0,
“lastOffloadLedgerId” : 0,
“lastOffloadSuccessTimeStamp” : 0,
“lastOffloadFailureTimeStamp” : 0,
“ongoingTxnCount” : 0,
“abortedTxnCount” : 0,
“committedTxnCount” : 0,
“publishers” : [ {
“accessMode” : “Shared”,
“msgRateIn” : 0.0,
“msgThroughputIn” : 0.0,
“averageMsgSize” : 0.0,
“chunkedMessageRate” : 0.0,
“producerId” : 3,
“supportsPartialProducer” : false,
“producerName” : “pulsar-0-106",
“address” : “/10.0.4.4:48576",
“connectedSince” : “2023-09-21T11:28:28.642147948Z”,
“clientVersion” : “2.10.4",
“metadata” : { }
} ],
“waitingPublishers” : 0,
“subscriptions” : {
“__compaction” : {
“msgRateOut” : 0.0,
“msgThroughputOut” : 0.0,
“bytesOutCounter” : 0,
“msgOutCounter” : 0,
“msgRateRedeliver” : 0.0,
“messageAckRate” : 0.0,
“chunkedMessageRate” : 0,
“msgBacklog” : 184244,
“backlogSize” : 4842296220,
“earliestMsgPublishTimeInBacklog” : 0,
“msgBacklogNoDelayed” : 184244,
“blockedSubscriptionOnUnackedMsgs” : false,
“msgDelayed” : 0,
“unackedMessages” : 0,
“type” : “None”,
“msgRateExpired” : 0.0,
“totalMsgExpired” : 0,
“lastExpireTimestamp” : 0,
“lastConsumedFlowTimestamp” : 0,
“lastConsumedTimestamp” : 0,
“lastAckedTimestamp” : 0,
“lastMarkDeleteAdvancedTimestamp” : 0,
“consumers” : [ ],
“isDurable” : true,
“isReplicated” : false,
“allowOutOfOrderDelivery” : false,
“consumersAfterMarkDeletePosition” : { },
“nonContiguousDeletedMessagesRanges” : 0,
“nonContiguousDeletedMessagesRangesSerializedSize” : 0,
“delayedMessageIndexSizeInBytes” : 0,
“subscriptionProperties” : { },
“filterProcessedMsgCount” : 0,
“filterAcceptedMsgCount” : 0,
“filterRejectedMsgCount” : 0,
“filterRescheduledMsgCount” : 0,
“durable” : true,
“replicated” : false
},
“websight-dxp-experiences-search-index-547cdf9c5d-dxv7v” : {
“msgRateOut” : 1009.3990130769382,
“msgThroughputOut” : 1.5503941574612858E7,
“bytesOutCounter” : 4276153214,
“msgOutCounter” : 278397,
“msgRateRedeliver” : 0.0,
“messageAckRate” : 168.03316794388724,
“chunkedMessageRate” : 0,
“msgBacklog” : 17337,
“backlogSize” : 355410316,
“earliestMsgPublishTimeInBacklog” : 0,
“msgBacklogNoDelayed” : 17337,
“blockedSubscriptionOnUnackedMsgs” : false,
“msgDelayed” : 0,
“unackedMessages” : 0,
“type” : “Exclusive”,
“activeConsumerName” : “b3250”,
“msgRateExpired” : 0.0,
“totalMsgExpired” : 0,
“lastExpireTimestamp” : 0,
“lastConsumedFlowTimestamp” : 1695295988291,
“lastConsumedTimestamp” : 1695295988263,
“lastAckedTimestamp” : 1695295988270,
“lastMarkDeleteAdvancedTimestamp” : 0,
“consumers” : [ {
“msgRateOut” : 1009.3990130769382,
“msgThroughputOut” : 1.5503941574612858E7,
“bytesOutCounter” : 4276153214,
“msgOutCounter” : 278397,
“msgRateRedeliver” : 0.0,
“messageAckRate” : 168.03316794388724,
“chunkedMessageRate” : 0.0,
“consumerName” : “b3250",
“availablePermits” : 21,
“unackedMessages” : 0,
“avgMessagesPerEntry” : 1,
“blockedConsumerOnUnackedMsgs” : false,
“address” : “/10.0.6.4:43380",
“connectedSince” : “2023-09-21T11:28:28.616981177Z”,
“clientVersion” : “Pulsar-Java-v3.1.0",
“lastAckedTimestamp” : 1695295988270,
“lastConsumedTimestamp” : 1695295988263,
“lastConsumedFlowTimestamp” : 1695295988291,
“metadata” : { },
“lastAckedTime” : “2023-09-21T11:33:08.27Z”,
“lastConsumedTime” : “2023-09-21T11:33:08.263Z”
} ],
“isDurable” : true,
“isReplicated” : false,
“allowOutOfOrderDelivery” : false,
“consumersAfterMarkDeletePosition” : { },
“nonContiguousDeletedMessagesRanges” : 0,
“nonContiguousDeletedMessagesRangesSerializedSize” : 0,
“delayedMessageIndexSizeInBytes” : 0,
“subscriptionProperties” : { },
“filterProcessedMsgCount” : 0,
“filterAcceptedMsgCount” : 0,
“filterRejectedMsgCount” : 0,
“filterRescheduledMsgCount” : 0,
“durable” : true,
“replicated” : false
}
},
“replication” : { },
“deduplicationStatus” : “Disabled”,
“nonContiguousDeletedMessagesRanges” : 0,
“nonContiguousDeletedMessagesRangesSerializedSize” : 0,
“delayedMessageIndexSizeInBytes” : 0,
“compaction” : {
“lastCompactionRemovedEventCount” : 0,
“lastCompactionSucceedTimestamp” : 0,
“lastCompactionFailedTimestamp” : 0,
“lastCompactionDurationTimeInMills” : 0
},
“ownerBroker” : “pulsar-broker-2.pulsar-broker.pulsar.svc.cluster.local:8080”
}
➜ ~ kubectl exec --namespace pulsar pulsar-toolset-0 -- bin/pulsar-admin topics stats-internal persistent://websight/dxp/search_experiences-partition-3
{
“entriesAddedCounter” : 0,
“numberOfEntries” : 184245,
“totalSize” : 4842312348,
“currentLedgerEntries” : 0,
“currentLedgerSize” : 0,
“lastLedgerCreatedTimestamp” : “2023-09-21T11:28:28.26Z”,
“waitingCursorsCount” : 0,
“pendingAddEntriesCount” : 0,
“lastConfirmedEntry” : “141:16664",
“state” : “LedgerOpened”,
“ledgers” : [ {
“ledgerId” : 35,
“entries” : 50000,
“size” : 806438303,
“offloaded” : false,
“underReplicated” : false
}, {
“ledgerId” : 83,
“entries” : 50000,
“size” : 806539134,
“offloaded” : false,
“underReplicated” : false
}, {
“ledgerId” : 89,
“entries” : 50000,
“size” : 806515803,
“offloaded” : false,
“underReplicated” : false
}, {
“ledgerId” : 126,
“entries” : 17580,
“size” : 2149590472,
“offloaded” : false,
“underReplicated” : false
}, {
“ledgerId” : 141,
“entries” : 16665,
“size” : 273228636,
“offloaded” : false,
“underReplicated” : false
}, {
“ledgerId” : 165,
“entries” : 0,
“size” : 0,
“offloaded” : false,
“underReplicated” : false
} ],
“cursors” : {
“__compaction” : {
“markDeletePosition” : “35:0”,
“readPosition” : “35:1”,
“waitingReadOp” : false,
“pendingReadOps” : 0,
“messagesConsumedCounter” : -184244,
“cursorLedger” : -1,
“cursorLedgerLastEntry” : -1,
“individuallyDeletedMessages” : “[]“,
“lastLedgerSwitchTimestamp” : “2023-09-21T11:28:28.353Z”,
“state” : “NoLedger”,
“active” : false,
“numberOfEntriesSinceFirstNotAckedMessage” : 1,
“totalNonContiguousDeletedMessagesRange” : 0,
“subscriptionHavePendingRead” : false,
“subscriptionHavePendingReplayRead” : false,
“properties” : {
“CompactedTopicLedger” : 44
}
},
“websight-dxp-experiences-search-index-547cdf9c5d-dxv7v” : {
“markDeletePosition” : “126:16907",
“readPosition” : “35:1",
“waitingReadOp” : false,
“pendingReadOps” : 0,
“messagesConsumedCounter” : -17337,
“cursorLedger” : -1,
“cursorLedgerLastEntry” : -1,
“individuallyDeletedMessages” : “[]“,
“lastLedgerSwitchTimestamp” : “2023-09-21T11:28:28.353Z”,
“state” : “NoLedger”,
“active” : false,
“numberOfEntriesSinceFirstNotAckedMessage” : 0,
“totalNonContiguousDeletedMessagesRange” : 0,
“subscriptionHavePendingRead” : false,
“subscriptionHavePendingReplayRead” : false,
“properties” : { }
}
},
“schemaLedgers” : [ ],
“compactedLedger” : {
“ledgerId” : 44,
“entries” : 1,
“size” : 186,
“offloaded” : false,
“underReplicated” : false
}
}

### Are you willing to submit a PR?

- [ ] I'm willing to submit a PR!