fix concurrent error cause stuck in TransactionBufferHandlerImpl#endTxn

fanjianye · fanjianye · commit e5428b9c761d · 2024-11-06T17:38:20.000+08:00
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferHandlerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferHandlerImpl.java
@@ -128,6 +128,9 @@ private boolean checkRequestCredits(OpRequestSend op) {
             }
         } else {
             pendingRequests.add(op);
+            if (currentPermits != REQUEST_CREDITS_UPDATER.get(this)) {
+                checkPendingRequests();
+            }
             return false;
         }
     }
@@ -251,6 +254,7 @@ private void checkPendingRequests() {
                             invalid.recycle();
                         }
                         endTxn(polled);
+                        break;
                     } else {
                         REQUEST_CREDITS_UPDATER.incrementAndGet(this);
                     }

Original file line number	Diff line number	Diff line change
`@@ -128,6 +128,9 @@ private boolean checkRequestCredits(OpRequestSend op) {`
`128`	`128`	`}`
`129`	`129`	`} else {`
`130`	`130`	`pendingRequests.add(op);`
	`131`	`+ if (currentPermits != REQUEST_CREDITS_UPDATER.get(this)) {`
	`132`	`+ checkPendingRequests();`
	`133`	`+ }`
`131`	`134`	`return false;`
`132`	`135`	`}`
`133`	`136`	`}`
`@@ -251,6 +254,7 @@ private void checkPendingRequests() {`
`251`	`254`	`invalid.recycle();`
`252`	`255`	`}`
`253`	`256`	`endTxn(polled);`
	`257`	`+ break;`
`254`	`258`	`} else {`
`255`	`259`	`REQUEST_CREDITS_UPDATER.incrementAndGet(this);`
`256`	`260`	`}`