ThreadPoolTaskExecutor is not fully utilised

[otbe]

Hi there,

its me again and I have another question. Let's say we have two queues, two message listener containers and a shared ThreadPoolTaskExecutor. Something similar to this

@Bean("myTpe")
public ThreadPoolTaskExecutor executor() {
  final var taskExecutor = new ThreadPoolTaskExecutor();
  taskExecutor.setQueueCapacity(0);
  taskExecutor.setMaxPoolSize(120);
  taskExecutor.setCorePoolSize(120);
  // ...
  return taskExecutor;
}

@Bean(name = "containerOne")
public MessageListenerContainer<Object> containerOne(SqsAsyncClient sqsAsyncClient,  MessageListener<Object> listener, ThreadPoolTaskExecutor taskExecutor) {
  return SqsMessageListenerContainer.builder()
      .id("containerOne")
      .sqsAsyncClient(sqsAsyncClient)
      .messageListener(listener)
      .configure(
          c ->
              c.maxConcurrentMessages(60)
                  .componentsTaskExecutor(taskExecutor))
      .queueNames("queue1")
      .build();
}

@Bean(name = "containerTwo")
public MessageListenerContainer<Object> containerTwo(SqsAsyncClient sqsAsyncClient,  MessageListener<Object> listener, ThreadPoolTaskExecutor taskExecutor) {
  return SqsMessageListenerContainer.builder()
      .id("containerTwo")
      .sqsAsyncClient(sqsAsyncClient)
      .messageListener(listener)
      .configure(
          c ->
              c.maxConcurrentMessages(60)
                  .componentsTaskExecutor(taskExecutor))
      .queueNames("queue2")
      .build();
}

I would've expected to see at max 120 threads being used to process 60 messages from each queue. But this is not what is happening. Most of the time the overall active tasks in the ThreadPoolTaskExecutor is 60 (sometimes 61, 62 but not higher). Both queues receive a large amount of messages (~2k) basically at the same the time and my assumption is now that we process 60 messages (in 60 threads) from both queues.
I already played around with a lot of settings like the back pressure mode (setting it to always high throughput) but nothing really helps.

If I make it 80/40 in terms of max concurrent messages, 80 threads are occupied. If I make it 40/40, 40 threads are occupied. If I add more instances of my application then all of them have the same amount of threads being used (so there are really messages left in the queue to be processed). I feel like I hit some hidden limitation like there's only one max concurrent messages number applied for all containers.

Where is my conceptional misunderstanding of how things work?

[tomazfernandes]

Hey @otbe, welcome back.

I feel like I hit some hidden limitation

It's hard to say without a sample project reproducing the scenario, but most likely you have a bottleneck somewhere in your flow, the app is processing the messages faster than more messages are being delivered, and thus never gets over 60 / 80.

Some usual suspects:

• SqsAsyncClient

SqsAsyncClient
   .builder()
   .httpClientBuilder(NettyNioAsyncHttpClient.builder()
       .maxConcurrency(1000))
       .build()

• pollTimeout
• Localstack throughput might be a bottleneck if you're running against it
• Network latency might be a bottleneck if you're running locally against AWS
• If you're sending messages while consuming

Let me know if that helps.