QPIDJMS-552 just process single completion task on processCompletions

apache · Nov 2, 2021 · aba2e7f · aba2e7f
1 parent 89c026a
commit aba2e7f
Showing 1 changed file with 19 additions and 28 deletions.
diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/JmsSession.java
@@ -199,38 +199,29 @@ public void onPendingFailure(ProviderException cause) {
     }
 
     private void processCompletions() {
-        for (; ; ) {
-            assert processCompletion.get();
-            completionThread = Thread.currentThread();
-            try {
-                Runnable completionTask;
-                while ((completionTask = completionTasks.poll()) != null) {
-                    try {
-                        completionTask.run();
-                    } catch (Throwable t) {
-                        LOG.debug("errored on processCompletions duty cycle", t);
-                    }
+        assert processCompletion.get();
+        completionThread = Thread.currentThread();
+        try {
+            final Runnable completionTask = completionTasks.poll();
+            if (completionTask != null) {
+                try {
+                    completionTask.run();
+                } catch (Throwable t) {
+                    LOG.debug("errored on processCompletions duty cycle", t);
                 }
-            } finally {
-                completionThread = null;
-                processCompletion.set(false);
-            }
-            if (completionTasks.isEmpty()) {
-                return;
             }
-            // a racing asyncProcessCompletion has won: no need to fire a continuation
-            if (!processCompletion.compareAndSet(false, true)) {
-                return;
-            }
-            if (connection.getCompletionExecutorService() == null) {
-                // an exclusive per-session executor doesn't need to guarantee fair completion processing
-                // and can just keep on processing completion on this same thread
-                continue;
-            }
-            // guarantees completions belonging to other sessions to be processed
-            getCompletionExecutor().execute(this::processCompletions);
+        } finally {
+            completionThread = null;
+            processCompletion.set(false);
+        }
+        if (completionTasks.isEmpty()) {
+            return;
+        }
+        // a racing asyncProcessCompletion has won: no need to fire a continuation
+        if (!processCompletion.compareAndSet(false, true)) {
             return;
         }
+        getCompletionExecutor().execute(this::processCompletions);
     }
 
     private void asyncProcessCompletion(final Runnable completionTask) {