Retry to initialize the token store on exception for PSEP.

AxonFramework · Jul 7, 2022 · 609d63c · 609d63c
1 parent c636717
commit 609d63c
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 0 deletions.
diff --git a/...g/src/main/java/org/axonframework/eventhandling/pooled/PooledStreamingEventProcessor.java b/...g/src/main/java/org/axonframework/eventhandling/pooled/PooledStreamingEventProcessor.java
@@ -205,6 +205,7 @@ private void initializeTokenStore() {
             } catch (Exception e) {
                 logger.info("Error while initializing the Token Store. " +
                                     "This may simply indicate concurrent attempts to initialize.", e);
+                initializeTokenStore();
             }
         });
     }

diff --git a/...c/test/java/org/axonframework/eventhandling/pooled/PooledStreamingEventProcessorTest.java b/...c/test/java/org/axonframework/eventhandling/pooled/PooledStreamingEventProcessorTest.java
@@ -141,6 +141,36 @@ void testStartContinuesWhenTokenInitializationFails() {
         assertTrue(testSubject.isRunning());
     }
 
+    @Test
+    void testRetriesWhenTokenInitializationInitiallyFails() {
+        InMemoryTokenStore spy = spy(tokenStore);
+        setTestSubject(createTestSubject(b -> b.tokenStore(spy)));
+
+        doThrow(new RuntimeException("Simulated failure")).doCallRealMethod()
+                                                          .when(spy)
+                                                          .initializeTokenSegments(any(), anyInt(), any());
+
+        List<EventMessage<Integer>> events = IntStream.range(0, 100)
+                                                      .mapToObj(GenericEventMessage::new)
+                                                      .collect(Collectors.toList());
+        events.forEach(stubMessageSource::publishMessage);
+        mockEventHandlerInvoker();
+
+        testSubject.start();
+
+        assertTrue(testSubject.isRunning());
+
+        assertWithin(1, TimeUnit.SECONDS, () -> assertEquals(8, testSubject.processingStatus().size()));
+        assertWithin(2, TimeUnit.SECONDS, () -> {
+            long nonNullTokens = IntStream.range(0, 8)
+                                          .mapToObj(i -> tokenStore.fetchToken(PROCESSOR_NAME, i))
+                                          .filter(Objects::nonNull)
+                                          .count();
+            assertEquals(8, nonNullTokens);
+        });
+        assertEquals(8, testSubject.processingStatus().size());
+    }
+
     @Test
     void testStartShutsDownImmediatelyIfCoordinatorExecutorThrowsAnException() {
         ScheduledExecutorService spiedCoordinatorExecutor = spy(coordinatorExecutor);