QueueConsumer.java

public QueueConsumer(
		QueueConsumerModule<?> queueConsumerModule,
		RetryPolicy retryPolicy,
		PlatformTransactionManager transactionManager,
		int polledItemsLimit,
		long pollingPeriodInSecs,
		int processingThreadCount
) {
  // ...
	this.processingExecutorService = Executors.newFixedThreadPool(processingThreadCount);
}
  
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());
      List<Callable<Object>> itemProcessingCallables = constructItemProcessingCallables(itemIds);
      this.processingExecutorService.invokeAll(itemProcessingCallables);
    }
  } catch (Throwable th) {
    logger.error("Error while fetching queued items: " + th.getMessage(), th);
  }
}

private List<Callable<Object>> constructItemProcessingCallables(List<?> itemIds) {
  return itemIds.stream()
      .map(itemId -> Executors.callable(() -> processItemAndHandleErrorIfRaised(itemId)))
      .collect(Collectors.toList());
}