[pulsar] branch branch-2.9 updated: [fix][txn]: fix transaction log recover throw cursor already close (#14810)

[bogong]

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

bogong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.9 by this push:
     new 199b2d2  [fix][txn]: fix transaction log recover throw cursor already 
close (#14810)
199b2d2 is described below

commit 199b2d22e5e9b495e7ba2579a7ec3bf3d475e387
Author: congbo <39078850+congbobo...@users.noreply.github.com>
AuthorDate: Thu Mar 24 13:08:36 2022 +0800

    [fix][txn]: fix transaction log recover throw cursor already close (#14810)
    
    ### Motivation
    When Transactionlog recover fail throw CursorAlreadyClosedException, we 
should stop the recover op. the cursor was been closed, the transaction log was 
been closed, so we should stop the recover op, in order to release thread 
resources
    like https://github.com/apache/pulsar/pull/14781
    
    ### Modifications
    When recover fail by CursorAlreadyClosedException, comeplete recover
    
    (cherry picked from commit a14a97e9aea230adf0ce85224d8d47a1c5e0fb22)
---
 .../transaction/buffer/impl/TopicTransactionBuffer.java     |  3 ++-
 .../apache/pulsar/broker/transaction/TransactionTest.java   | 13 +++++++++++++
 .../transaction/coordinator/impl/MLTransactionLogImpl.java  |  4 ++--
 3 files changed, 17 insertions(+), 3 deletions(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
index 3b28966d..f108a16 100644
--- 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
+++ 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
@@ -710,9 +710,10 @@ public class TopicTransactionBuffer extends 
TopicTransactionBufferState implemen
                     && exception instanceof 
ManagedLedgerException.NonRecoverableLedgerException
                     || exception instanceof 
ManagedLedgerException.ManagedLedgerFencedException) {
                 isReadable = false;
+            } else {
+                outstandingReadsRequests.decrementAndGet();
             }
             recover.callBackException(exception);
-            outstandingReadsRequests.decrementAndGet();
         }
     }
 }
diff --git 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
index 2b1a2f4..a5a39ab 100644
--- 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
+++ 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
@@ -682,6 +682,19 @@ public class TransactionTest extends TransactionTestBase {
                         mlTransactionSequenceIdGenerator);
         Awaitility.await().untilAsserted(() ->
                 assertEquals(metadataStore2.getCoordinatorStats().state, 
"Ready"));
+
+        doAnswer(invocation -> {
+            AsyncCallbacks.ReadEntriesCallback callback = 
invocation.getArgument(1);
+            callback.readEntriesFailed(new 
ManagedLedgerException.CursorAlreadyClosedException("test"), null);
+            return null;
+        }).when(managedCursor).asyncReadEntries(anyInt(), any(), any(), any());
+
+        MLTransactionMetadataStore metadataStore3 =
+                new MLTransactionMetadataStore(new TransactionCoordinatorID(1),
+                        mlTransactionLog, timeoutTracker, 
transactionRecoverTracker,
+                        mlTransactionSequenceIdGenerator);
+        Awaitility.await().untilAsserted(() ->
+                assertEquals(metadataStore3.getCoordinatorStats().state, 
"Ready"));
     }
 
     @Test
diff --git 
a/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImpl.java
 
b/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImpl.java
index 9ab4bd3..2a13f95 100644
--- 
a/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImpl.java
+++ 
b/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImpl.java
@@ -271,11 +271,11 @@ public class MLTransactionLogImpl implements 
TransactionLog {
         public void readEntriesFailed(ManagedLedgerException exception, Object 
ctx) {
             if (managedLedgerConfig.isAutoSkipNonRecoverableData()
                     && exception instanceof 
ManagedLedgerException.NonRecoverableLedgerException
-                    || exception instanceof 
ManagedLedgerException.ManagedLedgerFencedException) {
+                    || exception instanceof 
ManagedLedgerException.ManagedLedgerFencedException
+                    || exception instanceof 
ManagedLedgerException.CursorAlreadyClosedException) {
                 isReadable = false;
             }
             log.error("Transaction log init fail error!", exception);
-            outstandingReadsRequests.decrementAndGet();
         }
 
     }