Merge pull request #2292 from nils-christian/2290

[#2290] `TrackingEventProcessor` does not wait for his worker threads to shut down
AxonFramework · Jul 20, 2022 · ef2df7b · ef2df7b
2 parents 8e15305 + 940fb17
commit ef2df7b
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 6 deletions.
diff --git a/messaging/src/main/java/org/axonframework/eventhandling/TrackingEventProcessor.java b/messaging/src/main/java/org/axonframework/eventhandling/TrackingEventProcessor.java
@@ -350,8 +350,13 @@ private void releaseToken(Segment segment) {
         try {
             transactionManager.executeInTransaction(() -> tokenStore.releaseClaim(getName(), segment.getSegmentId()));
             logger.info("Released claim");
-        } catch (Exception e) {
-            logger.info("Release claim failed", e);
+        } catch (Exception e) {  
+            // Ignore exception
+            if (logger.isDebugEnabled()) {
+                logger.debug("Release claim failed", e);
+            } else if (logger.isInfoEnabled()) {
+                logger.info("Release claim failed");
+            }
             // Ignore exception
         }
     }
@@ -696,6 +701,13 @@ private CompletableFuture<Void> awaitTermination() {
                                     if (workerThread.isAlive()) {
                                         workerThread.interrupt();
                                         workerThread.join(workerTerminationTimeout);
+                                        if (workerThread.isAlive()) {
+                                            logger.warn(
+                                                    "Forced shutdown of Tracking Processor Worker '{}' was unsuccessful. " 
+                                                            + "Consider increasing workerTerminationTimeout.",
+                                                    worker.getKey()
+                                            );
+                                        }
                                     }
                                 } catch (InterruptedException e) {
                                     logger.info(

diff --git a/...ng/src/main/java/org/axonframework/eventhandling/TrackingEventProcessorConfiguration.java b/...ng/src/main/java/org/axonframework/eventhandling/TrackingEventProcessorConfiguration.java
@@ -186,14 +186,32 @@ public TrackingEventProcessorConfiguration andEventTrackerStatusChangeListener(
     }
 
     /**
-     * Sets the shutdown timeout to terminate active workers. Defaults to 5000ms.
+     * Sets the shutdown timeout to terminate active workers. This is used for both the graceful termination and the 
+     * potential forced termination of active workers. It is thus possible that it is used twice during the shutdown 
+     * phase. Defaults to 5000ms.
      *
-     * @param workerTerminationTimeout the timeout for workers to terminate on a shutdown
+     * @param workerTerminationTimeout the timeout for workers to terminate on a shutdown in milliseconds
      * @return {@code this} for method chaining
+     * 
+     * @deprecated Use {@link #andWorkerTerminationTimeout(long, TimeUnit)} instead.
      */
-    public TrackingEventProcessorConfiguration andWorkerTerminationTimeout(long workerTerminationTimeout) {
+    @Deprecated
+    public TrackingEventProcessorConfiguration andWorkerTerminationTimeout(long workerTerminationTimeoutInMilliseconds) {
+    	return andWorkerTerminationTimeout(workerTerminationTimeoutInMilliseconds, TimeUnit.MILLISECONDS);
+    }
+
+    /**
+     * Sets the shutdown timeout to terminate active workers. This is used for both the graceful termination and the 
+     * potential forced termination of active workers. It is thus possible that it is used twice during the shutdown 
+     * phase. Defaults to 5000ms.
+     *
+     * @param workerTerminationTimeout the timeout for workers to terminate on a shutdown.
+     * @param timeUnit           The unit of time
+     * @return {@code this} for method chaining
+     */
+    public TrackingEventProcessorConfiguration andWorkerTerminationTimeout(long workerTerminationTimeout, TimeUnit timeUnit) {
         assertStrictPositive(workerTerminationTimeout, "The worker termination timeout should be strictly positive");
-        this.workerTerminationTimeout = workerTerminationTimeout;
+        this.workerTerminationTimeout = timeUnit.toMillis(workerTerminationTimeout);
         return this;
     }