[pulsar] branch master updated: [fix][broker] fix ttl expiration block due to no-recoverable exception even if autoSkipNonRecoverableData=true (#19132)

[aloyszhang]

This is an automated email from the ASF dual-hosted git repository.

aloyszhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new badd69b8e01 [fix][broker] fix ttl expiration block due to 
no-recoverable exception even if autoSkipNonRecoverableData=true (#19132)
badd69b8e01 is described below

commit badd69b8e012162feb28c86ca9de527cbfd1ac22
Author: AloysZhang <lofterzh...@gmail.com>
AuthorDate: Sun Jan 8 20:23:55 2023 +0800

    [fix][broker] fix ttl expiration block due to no-recoverable exception even 
if autoSkipNonRecoverableData=true (#19132)
---
 .../pulsar/broker/service/persistent/PersistentTopic.java      | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 65831d996c4..15bf568cbb2 100644
--- 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -67,6 +67,7 @@ import 
org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerAlready
 import 
org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerFencedException;
 import 
org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerTerminatedException;
 import 
org.apache.bookkeeper.mledger.ManagedLedgerException.MetadataNotFoundException;
+import 
org.apache.bookkeeper.mledger.ManagedLedgerException.NonRecoverableLedgerException;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.impl.ManagedCursorContainer;
 import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
@@ -2858,7 +2859,14 @@ public class PersistentTopic extends AbstractTopic 
implements Topic, AddEntryCal
                         (int) (messageTTLInSeconds * 
MESSAGE_EXPIRY_THRESHOLD), entryTimestamp);
             }
         } catch (Exception e) {
-            log.warn("[{}] Error while getting the oldest message", topic, e);
+            if 
(brokerService.pulsar().getConfiguration().isAutoSkipNonRecoverableData()
+                    && e instanceof NonRecoverableLedgerException) {
+                // NonRecoverableLedgerException means the ledger or entry 
can't be read anymore.
+                // if AutoSkipNonRecoverableData is set to true, just return 
true here.
+                return true;
+            } else {
+                log.warn("[{}] Error while getting the oldest message", topic, 
e);
+            }
         } finally {
             if (entry != null) {
                 entry.release();