QueueConsumer.java

public QueueConsumer(
    QueueConsumerModule<?> queueConsumerModule,
    RetryPolicy retryPolicy,
    PlatformTransactionManager transactionManager,
    int polledItemsLimit,
    long pollingPeriodInSecs,
    int partitionCount
) {
// ...
  this.scheduledExecutorService = Executors.newScheduledThreadPool(partitionCount);
}

// ...

private void startProcessingTasks() {
  logger.info("Starting {} queue polling tasks with delay of {} secs", partitionCount, pollingPeriodInSecs);

  Set<ScheduledFuture<?>> tasks = new HashSet<>();
  for (int i = 0; i < partitionCount; i++) {
    final int partition = i; // lambda requires 'final' variable, and loop requires it to be incremented, so we have to do it this way
    Runnable command = () -> processQueuedItems(partition);
    ScheduledFuture<?> task = this.scheduledExecutorService.scheduleWithFixedDelay(command, pollingPeriodInSecs, pollingPeriodInSecs, TimeUnit.SECONDS);
    tasks.add(task);
  }

  this.processingTasks = tasks;
}