fix concurrent error cause stuck in TransactionBufferHandlerImpl#endTxn

Author: fanjianye

Diff for: 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;
         }
     }
@@ -237,26 +240,25 @@ public void onResponse(OpRequestSend op) {
     }
 
     private void checkPendingRequests() {
-        while (true) {
-            int permits = REQUEST_CREDITS_UPDATER.get(this);
-            if (permits > 0 && pendingRequests.peek() != null) {
-                if (REQUEST_CREDITS_UPDATER.compareAndSet(this, permits, permits - 1)) {
-                    OpRequestSend polled = pendingRequests.poll();
-                    if (polled != null) {
-                        CompletableFuture<ClientCnx> clientCnx = getClientCnx(polled.topic);
-                        if (polled.cnx != clientCnx) {
-                            OpRequestSend invalid = polled;
-                            polled = OpRequestSend.create(invalid.requestId, invalid.topic, invalid.cmd, invalid.cb,
-                                    clientCnx);
-                            invalid.recycle();
-                        }
-                        endTxn(polled);
-                    } else {
-                        REQUEST_CREDITS_UPDATER.incrementAndGet(this);
+        int permits = REQUEST_CREDITS_UPDATER.get(this);
+        if (permits > 0 && pendingRequests.peek() != null) {
+            if (REQUEST_CREDITS_UPDATER.compareAndSet(this, permits, permits - 1)) {
+                OpRequestSend polled = pendingRequests.poll();
+                if (polled != null) {
+                    CompletableFuture<ClientCnx> clientCnx = getClientCnx(polled.topic);
+                    if (polled.cnx != clientCnx) {
+                        OpRequestSend invalid = polled;
+                        polled = OpRequestSend.create(invalid.requestId, invalid.topic, invalid.cmd, invalid.cb,
+                                clientCnx);
+                        invalid.recycle();
                     }
+                    endTxn(polled);
+                } else {
+                    REQUEST_CREDITS_UPDATER.incrementAndGet(this);
+                    checkPendingRequests();
                 }
             } else {
-                break;
+                checkPendingRequests();
             }
         }
     }