title: How to process more than 10 concurrent messages from an AWS SQS FiFo queue using Spring Integration
body: <p>I want to be able to process more than 10 SQS messages at a time using a Spring Integration Workflow.</p>
  
  <p>From this question, the recommendation was to use an <strong>ExecutorChannel</strong>. I updated my code but still have the same symptoms. </p>
  
  <p><a href="https://stackoverflow.com/questions/49802891/how-execute-spring-integration-flow-in-multiple-threads-to-consume-more-amazon-s">How execute Spring integration flow in multiple threads to consume more Amazon SQS queue messages in parallel?</a></p>
  
  <p>After making this update, my application requests 10 messages, processes those, and only after I make the call to <strong>amazonSQSClient.deleteMessage</strong> near the end of the flow will it accept another 10 messages from the SQS queue.</p>
  
  <p>The application uses an SQS FiFo queue.</p>
  
  <p>Is there something else I'm missing, or is this an unavoidable symptom of using <strong>SqsMessageDeletionPolicy.NEVER</strong> and then deleting the messages at the end of the flow? Accepting the messages at the beginning of the flow isn't really an option due to other constraints.</p>
  
  <p>Here are the relevant snippets of code, with some simplifications, but I hope it expresses the problem. </p>
  
  <p><strong>Queue configuration</strong></p>
  
  <pre><code>@Bean
  public AsyncTaskExecutor inputChannelTaskExecutor() {
      SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
      executor.setConcurrencyLimit(50);
      return executor;
  }
  
  @Bean
  @Qualifier("inputChannel")
  public ExecutorChannel inputChannel() {
      return new ExecutorChannel(inputChannelTaskExecutor());
  }
  </code></pre>
  
  <p>I also tried a ThreadPoolTaskExecutor instead of the SimpleAsyncTaskExecutor, with the same result but I'll include that too, in case it offers other insight.</p>
  
  <pre><code>    @Bean
      public AsyncTaskExecutor inputChannelTaskExecutor() {
          ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
          executor.setMaxPoolSize(50);
          executor.setQueueCapacity(50);
          executor.setThreadNamePrefix("spring-async-");
          executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
          executor.afterPropertiesSet();
          executor.initialize();
          return executor;
      }
  </code></pre>
  
  <p><strong>SQS Channel Adapter</strong></p>
  
  <pre><code>@Bean
  public SqsMessageDrivenChannelAdapter changeQueueMessageAdapter() {
      SqsMessageDrivenChannelAdapter adapter = new SqsMessageDrivenChannelAdapter(this.amazonSQSClient, changeQueue);
      adapter.setOutputChannel(inputChannel);
      adapter.setMessageDeletionPolicy(SqsMessageDeletionPolicy.NEVER);
      return adapter;
  }
  
  
  @Bean(name = PollerMetadata.DEFAULT_POLLER)
  public PollerSpec poller() {
      return Pollers.fixedRate(500, TimeUnit.MILLISECONDS).maxMessagesPerPoll(10);
  }
  </code></pre>
  
  <p><strong>Simplified main flow</strong></p>
  
  <p>A common scenario for us is to get a number of Branch edits in a short period of time. This flow only 'cares' that at least one edit has happened. The <strong>messageTransformer</strong> extracts an id from the payload document and puts it in the header <strong>dsp_docId</strong> which we then use to aggregate on (we use this id in a few other places, so we felt a header made sense rather than doing all the work in a custom aggregator).</p>
  
  <p>The <strong>provisioningServiceActivator</strong> retrieves the latest version of the Branch, then the router decides whether it needs further transforms (in which case it sends it to the <strong>transformBranchChannel</strong>) or it can be sent onto our PI instance (via the sendToPiChannel). </p>
  
  <p>The transform flow (not shown, I don't think you need it) leads to the sent to PI flow eventually, it just does more work first.</p>
  
  <p>The <strong>listingGroupProcessor</strong> captures all the <strong>aws_receiptHandle</strong> headers and adds them to a new header as a | separated list.</p>
  
  <p>The sendToPi flow (and the errorFlow) ends with a call to a custom handler that takes care of deleting all the SQS messages referred to by that list of aws_receiptHandle strings. </p>
  
  <pre><code>@Bean
  IntegrationFlow sqsListener() {
      return IntegrationFlows.from(inputChannel)
                             .transform(messageTransformer)
                             .aggregate(a -&gt; a.correlationExpression("1")
                                              .outputProcessor(listingGroupProcessor)
                                              .autoStartup(true)
                                              .correlationStrategy(message -&gt; message.getHeaders().get("dsp_docId"))
                                              .groupTimeout(messageAggregateTimeout)  // currently 25s
                                              .expireGroupsUponCompletion(true)
                                              .sendPartialResultOnExpiry(true)
                                              .get())
  
                             .handle(provisioningServiceActivator, "handleStandard")
                             .route(Branch.class, branch -&gt; (branch.isSuppressed() == null || !branch.isSuppressed()),
                                    routerSpec -&gt; routerSpec.channelMapping(true, "transformBranchChannel")
                                                            .resolutionRequired(false)
                                                            .defaultOutputToParentFlow())
  
                             .channel(sendtoPiChannel)
                             .get();
  }
  </code></pre>