MessageListenerThread extends Thread { void checkConsumerDelay(ActionLog actionLog, long timestamp, long longConsumerDelayThresholdInNano) { long delay = (actionLog.date.toEpochMilli() - timestamp) * 1_000_000; actionLog.stats.put("consumer_delay", (double) delay); if (delay > longConsumerDelayThresholdInNano) { logger.warn(errorCode("LONG_CONSUMER_DELAY"), "consumer delay is too long, delay={}", Duration.ofNanos(delay)); } } MessageListenerThread(String name, MessageListener listener); @Override void run();  }
@Test void checkConsumerDelay() { var actionLog = logManager.begin(null, null); thread.checkConsumerDelay(actionLog, actionLog.date.minusSeconds(5).toEpochMilli(), Duration.ofSeconds(3).toNanos()); assertThat(actionLog.stats).containsEntry("consumer_delay", (double) Duration.ofSeconds(5).toNanos()); assertThat(actionLog.errorCode()).isEqualTo("LONG_CONSUMER_DELAY"); actionLog = logManager.begin(null, null); thread.checkConsumerDelay(actionLog, actionLog.date.minusSeconds(1).toEpochMilli(), Duration.ofSeconds(6).toNanos()); assertThat(actionLog.stats).containsEntry("consumer_delay", (double) Duration.ofSeconds(1).toNanos()); assertThat(actionLog.errorCode()).isNull(); }