QueueConsumer.java

public void processQueuedItems() {
	try {
		LocalDateTime now = LocalDateTime.now();
		List<?> itemIds = this.queueConsumerModule.findItemIdsWhereQueueingNextAttemptTimeIsBefore(now, itemsPollSize);

		if (!itemIds.isEmpty()) {
			logger.info("Fetched {} pending queued items", itemIds.size());
			for (Object itemId : itemIds) {
				processItemAndHandleErrorIfRaised(itemId);
			}
		}
	} catch (Throwable th) {
		logger.error("Error while fetching queued items: " + th.getMessage(), th);
	}
}

private void processItemAndHandleErrorIfRaised(Object itemId) {
	try {
		executeUnderTransaction(() -> processItem(itemId));
	} catch (Throwable error) {
		executeUnderTransaction(() -> registerProcessingFailure(itemId, error));
	}
}