@Test void triggerReplayWillThrowIllegalStateException_WhenNoMatchingSwitchingEventProcessorFound() { when(eventProcessingConfiguration.eventProcessorByProcessingGroup( "foo", ReplayableEventProcessor.class)).thenReturn(Optional.empty()); assertThrows(IllegalStateException.class, () -> replayEndpoint.startReplay(Set.of("foo"), null)); }
@WriteOperation public void startReplay(@Nullable Set<String> processingGroups, @Nullable OffsetDateTime startTime) { var replayableEventProcessors = processingGroups == null ? getReplayableEventProcessors() : getReplayableEventProcessor(processingGroups); if (replayableEventProcessors.isEmpty()) { throw new IllegalStateException("No matching replayable event processors"); } synchronized (this) { if (isReplaying()) { throw new IllegalStateException("Cannot start replay while an existing one is running"); } EventStore eventStore = axonConfiguration.eventStore(); TrackingToken startPosition = startTime == null ? eventStore.createTailToken() : eventStore.createTokenAt(startTime.toInstant()); ReplayMarkerEvent replayMarkerEvent = new ReplayMarkerEvent(randomUUID()); replayableEventProcessors.forEach(p -> { replayingProcessors.put(p, p.registerReplayCompletionListener(this::onSingleProcessorReplayCompletion)); p.startReplay(startPosition, replayMarkerEvent); }); axonConfiguration.eventGateway().publish(replayMarkerEvent); } }
ReplayEndpoint { @WriteOperation public void startReplay(@Nullable Set<String> processingGroups, @Nullable OffsetDateTime startTime) { var replayableEventProcessors = processingGroups == null ? getReplayableEventProcessors() : getReplayableEventProcessor(processingGroups); if (replayableEventProcessors.isEmpty()) { throw new IllegalStateException("No matching replayable event processors"); } synchronized (this) { if (isReplaying()) { throw new IllegalStateException("Cannot start replay while an existing one is running"); } EventStore eventStore = axonConfiguration.eventStore(); TrackingToken startPosition = startTime == null ? eventStore.createTailToken() : eventStore.createTokenAt(startTime.toInstant()); ReplayMarkerEvent replayMarkerEvent = new ReplayMarkerEvent(randomUUID()); replayableEventProcessors.forEach(p -> { replayingProcessors.put(p, p.registerReplayCompletionListener(this::onSingleProcessorReplayCompletion)); p.startReplay(startPosition, replayMarkerEvent); }); axonConfiguration.eventGateway().publish(replayMarkerEvent); } } }
ReplayEndpoint { @WriteOperation public void startReplay(@Nullable Set<String> processingGroups, @Nullable OffsetDateTime startTime) { var replayableEventProcessors = processingGroups == null ? getReplayableEventProcessors() : getReplayableEventProcessor(processingGroups); if (replayableEventProcessors.isEmpty()) { throw new IllegalStateException("No matching replayable event processors"); } synchronized (this) { if (isReplaying()) { throw new IllegalStateException("Cannot start replay while an existing one is running"); } EventStore eventStore = axonConfiguration.eventStore(); TrackingToken startPosition = startTime == null ? eventStore.createTailToken() : eventStore.createTokenAt(startTime.toInstant()); ReplayMarkerEvent replayMarkerEvent = new ReplayMarkerEvent(randomUUID()); replayableEventProcessors.forEach(p -> { replayingProcessors.put(p, p.registerReplayCompletionListener(this::onSingleProcessorReplayCompletion)); p.startReplay(startPosition, replayMarkerEvent); }); axonConfiguration.eventGateway().publish(replayMarkerEvent); } } @Autowired  ReplayEndpoint(AxonConfiguration axonConfiguration,
                          List<ReplayCompletionAware> replayCompletionAwares,
                          TaskExecutor taskExecutor); }
ReplayEndpoint { @WriteOperation public void startReplay(@Nullable Set<String> processingGroups, @Nullable OffsetDateTime startTime) { var replayableEventProcessors = processingGroups == null ? getReplayableEventProcessors() : getReplayableEventProcessor(processingGroups); if (replayableEventProcessors.isEmpty()) { throw new IllegalStateException("No matching replayable event processors"); } synchronized (this) { if (isReplaying()) { throw new IllegalStateException("Cannot start replay while an existing one is running"); } EventStore eventStore = axonConfiguration.eventStore(); TrackingToken startPosition = startTime == null ? eventStore.createTailToken() : eventStore.createTokenAt(startTime.toInstant()); ReplayMarkerEvent replayMarkerEvent = new ReplayMarkerEvent(randomUUID()); replayableEventProcessors.forEach(p -> { replayingProcessors.put(p, p.registerReplayCompletionListener(this::onSingleProcessorReplayCompletion)); p.startReplay(startPosition, replayMarkerEvent); }); axonConfiguration.eventGateway().publish(replayMarkerEvent); } } @Autowired  ReplayEndpoint(AxonConfiguration axonConfiguration,
                          List<ReplayCompletionAware> replayCompletionAwares,
                          TaskExecutor taskExecutor); @ReadOperation Map<String, Object> status(); @WriteOperation void startReplay(@Nullable Set<String> processingGroups,
                            @Nullable OffsetDateTime startTime); }
ReplayEndpoint { @WriteOperation public void startReplay(@Nullable Set<String> processingGroups, @Nullable OffsetDateTime startTime) { var replayableEventProcessors = processingGroups == null ? getReplayableEventProcessors() : getReplayableEventProcessor(processingGroups); if (replayableEventProcessors.isEmpty()) { throw new IllegalStateException("No matching replayable event processors"); } synchronized (this) { if (isReplaying()) { throw new IllegalStateException("Cannot start replay while an existing one is running"); } EventStore eventStore = axonConfiguration.eventStore(); TrackingToken startPosition = startTime == null ? eventStore.createTailToken() : eventStore.createTokenAt(startTime.toInstant()); ReplayMarkerEvent replayMarkerEvent = new ReplayMarkerEvent(randomUUID()); replayableEventProcessors.forEach(p -> { replayingProcessors.put(p, p.registerReplayCompletionListener(this::onSingleProcessorReplayCompletion)); p.startReplay(startPosition, replayMarkerEvent); }); axonConfiguration.eventGateway().publish(replayMarkerEvent); } } @Autowired  ReplayEndpoint(AxonConfiguration axonConfiguration,
                          List<ReplayCompletionAware> replayCompletionAwares,
                          TaskExecutor taskExecutor); @ReadOperation Map<String, Object> status(); @WriteOperation void startReplay(@Nullable Set<String> processingGroups,
                            @Nullable OffsetDateTime startTime);  }