Merge pull request #2305 from AxonFramework/bug/2293

Fix a problem where when a shutdown takes places while the worklaunch…
AxonFramework · Jul 28, 2022 · c7fb25a · c7fb25a
2 parents 411ed1e + 80b6cba
commit c7fb25a
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 2 deletions.
diff --git a/...xonframework/integrationtests/eventhandling/TrackingEventProcessorTest_MultiThreaded.java b/...xonframework/integrationtests/eventhandling/TrackingEventProcessorTest_MultiThreaded.java
@@ -422,6 +422,21 @@ void testMultiThreadTokensAreStoredWhenUnitOfWorkIsRolledBackOnSecondEvent() thr
         assertNotNull(tokenStore.fetchToken(testSubject.getName(), 1));
     }
 
+    @Test
+    void testProcessorIncrementAndDecrementCorrectly() throws InterruptedException {
+        configureProcessor(TrackingEventProcessorConfiguration.forParallelProcessing(2)
+                                                              .andInitialSegmentsCount(4));
+        testSubject.start();
+        // It's an edge case, but locally this successfully fails with the previous implementation.
+        // It would before prevent to increase available threads because the launcher was running.
+        testSubject.shutDown();
+        testSubject.start();
+        testSubject.shutDown();
+        testSubject.start();
+        Thread.sleep(200);
+        assertEquals(2, testSubject.activeProcessorThreads());
+    }
+
     // Utility to add up acknowledged messages by Thread (worker) name and assertions facilities.
     class AcknowledgeByThread {
 

diff --git a/messaging/src/main/java/org/axonframework/eventhandling/TrackingEventProcessor.java b/messaging/src/main/java/org/axonframework/eventhandling/TrackingEventProcessor.java
@@ -20,6 +20,7 @@
 import org.axonframework.common.AxonConfigurationException;
 import org.axonframework.common.AxonNonTransientException;
 import org.axonframework.common.ExceptionUtils;
+import org.axonframework.common.ProcessUtils;
 import org.axonframework.common.stream.BlockingStream;
 import org.axonframework.common.transaction.NoTransactionManager;
 import org.axonframework.common.transaction.TransactionManager;
@@ -692,6 +693,7 @@ private CompletableFuture<Void> awaitTermination() {
         }
 
         logger.info("Processor '{}' awaiting termination...", getName());
+        ProcessUtils.executeUntilTrue(() -> !workLauncherRunning.get(), 10L, 100L);
         return workerThreads.entrySet()
                             .stream()
                             .map(worker -> CompletableFuture.runAsync(() -> {
@@ -1084,7 +1086,9 @@ public void cleanUp() {
             workerThreads.remove(name());
             logger.info("Worker for segment {} stopped.", segment);
 
-            if (!workLauncherRunning.get() && availableThreads.getAndIncrement() == 0 && getState().isRunning()) {
+            final int currentAvailableThreads = availableThreads.getAndIncrement();
+
+            if (!workLauncherRunning.get() && currentAvailableThreads == 0 && getState().isRunning()) {
                 logger.info("No Worker Launcher active. Using current thread to assign segments.");
                 new WorkerLauncher().run();
             }
@@ -1093,6 +1097,8 @@ public void cleanUp() {
 
     private class WorkerLauncher implements Worker {
 
+        private boolean asSegmentWorker = false;
+
         @Override
         public void run() {
             try {
@@ -1208,6 +1214,9 @@ segmentId, getName(), e
                     // We're not able to spawn new threads, so this thread should also start processing.
                     if (nonNull(workingInCurrentThread)) {
                         logger.info("Using current Thread for last segment worker: {}", workingInCurrentThread);
+                        workerThreads.remove(name());
+                        workerThreads.put(workingInCurrentThread.name(), Thread.currentThread());
+                        asSegmentWorker = true;
                         workLauncherRunning.set(false);
                         workingInCurrentThread.run();
                         return;
@@ -1227,7 +1236,9 @@ public String name() {
 
         @Override
         public void cleanUp() {
-            workerThreads.remove(name());
+            if (! asSegmentWorker){
+                workerThreads.remove(name());
+            }
         }
     }