TrackingProcessor processingStatus

Imported from Google Groups
#1

Hello,

I would like to implement a health status actuator to see the status of my trackingProcessors. It should become unhealthy when one of the tracking processors is running behind or stopped.
I know in the latest Axon versions, there is a processingStatus on TrackingProcessors.

How can I get access to all tracking processors? I cant inject them and are autocreated.

Kind regards,
Koen

#2

Hi Koen,

Perhaps I can help. I recently wrote a very rudimentary TrackingProcessorMonitor that would raise an event as soon as a tracking processor caught up.

public class TrackingProcessorMonitor {

   private EventHandlingConfiguration eventHandlingConfiguration;
   private EventBus eventBus;

   public TrackingProcessorMonitor(EventHandlingConfiguration eventHandlingConfiguration, EventBus eventBus) {
      this.eventHandlingConfiguration = eventHandlingConfiguration;
      this.eventBus = eventBus;
   }

   @EventListener
   public void init(ApplicationReadyEvent e) {
      final ExecutorService executorService = Executors.newFixedThreadPool(10);
      eventHandlingConfiguration.getProcessors().stream()
         .filter(p -> p instanceof TrackingEventProcessor).forEach(p -> executorService.submit(new Worker((TrackingEventProcessor) p, this.eventBus)));
   }

   @Slf4j
   private static class Worker implements Runnable {

      private TrackingEventProcessor processor;
      private EventBus eventBus;

      Worker(TrackingEventProcessor processor, EventBus eventBus) {
         this.processor = processor;
         this.eventBus = eventBus;
      }

      @Override
      public void run() {
         log.debug("Monitoring tracking processor '{}' initialization...", processor.getName());
         boolean caughtUp;
         do {
            caughtUp = this.processor.processingStatus().values().stream().allMatch(EventTrackerStatus::isCaughtUp);
            try {
               Thread.sleep(500);
            } catch (InterruptedException e) {
            }
         }
         while (!caughtUp);

         log.debug("Tracking processor '{}' caught-up", processor.getName());
         this.eventBus.publish(GenericEventMessage.asEventMessage(new ProcessorCaughtUpEvent(this.processor.getName())));
      }
   }

}

The line of interest for your use-case:

eventHandlingConfiguration.getProcessors()

Sounds like a status we’d also be interested in using so please keep us posted.

Regards,
Dylan

#3

Thanks, thats what I need ! :slight_smile:

#4

Notice that this probably won’t work when using distributed. The event processor thread(s) might live on another instance, and so the status will not be found on this one or is incomplete.

#5

Hi all,

I am stil looking for a good solution.
Any new features in Axon 4.2/4.3 that can help with this?

Kind regards
Koen

#6

Koen , can you say more about your configuration ?
If you are using axon server you could use the event processor event endpoint like :
http://axonserver:8024/components//processors?context=

To get all processors in an application you could use EventProcessingConfiguration (injectable) and call eventProcessors() on it .