title: Spring Kafka limiting number of messages to be consumed in a time period
body: <p>I have a Spring Boot application which listens a Kafka topic with Spring Kafka. After the message is consumed, several web/rest service calls are executed with the information inside the message to collect some other data and this process takes some time as expected. So, I used a thread pool with the size of 20 in order to create a parallel message processing.</p>
  <p>This system generally works well, but rarely large number of messages (around 200K) put/produced to the Kafka topic in a short period of time (1 second). At this situation, consumer consumes the messages immediately, but message processing mechanism is not fast enough. Therefore all the consumed messages stay in memory while waiting a thread and the application gets OutOfMemoryError.</p>
  <p>Increasing the thread pool size to the some point can be a improvement but it is not a permanent solution for this problem. I want to create a balance with the number of messages consumed and the number of messages processed in a period of time. This can be limiting the number of messages consumed from the Kafka topic or consuming a message when there is a possibility to process it immediately.</p>
  <p>Is there any Kafka consumer configuration to limit the number of messages <strong>in a period of time</strong>? How can I optimize the consuming and processing mechanism when the latency for the message consumption is not a problem?</p>
  <p>PS: It seems there is no configuration for the time interval between two subsequent polls (<a href="https://stackoverflow.com/questions/52655108/what-is-the-delay-time-between-each-poll">What is the delay time between each poll</a>), if it exists there may be a solution with that configuration.</p>
  <p>Here is my consumer code:</p>
  <pre><code>@Autowired
  MessageProcessUtil messageProcessUtil;
  
  private ExecutorService executor = Executors.newFixedThreadPool(20);
  
  @KafkaListener(topics = &quot;${kafka.consumer.topicName}&quot;)
  public void consume(String message){
      logger.info(String.format(&quot;$$ -&gt; Consumed Message -&gt; %s&quot;,message));
      messageProcessUtil.processMessage(message, executor);
  }
  </code></pre>
  <p>Consumer configuration:</p>
  <pre><code>kafka.consumer.enable.auto.commit=true
  kafka.consumer.auto.commit.interval.ms=1000
  kafka.consumer.request.timeout.ms=40000
  kafka.consumer.session.timeout.ms=30000
  kafka.consumer.max.poll.records=1
  kafka.consumer.fetch.max.wait.ms=500
  kafka.consumer.auto.offset.reset=earliest
  </code></pre>
  <p>Thanks in advance for your help.</p>