title: Spring kafka consume records with some delay
body: <p>I'm using spring kafka in my application. I want to add some delay of 15 mins for consuming the records for one of the listener - kafkaRetryListenerContainerFactory. I have two listeners. Below is my configuration:</p>
  
  <pre><code>@Bean
  public ConcurrentKafkaListenerContainerFactory&lt;String, Object&gt; kafkaListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory&lt;String, Object&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(primaryConsumerFactory());
      return factory;
  }
  
  @Bean
  public ConcurrentKafkaListenerContainerFactory&lt;String, Object&gt; kafkaRetryListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory&lt;String, Object&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(retryConsumerFactory());
      factory.setConcurrency(this.kafkaConfigProperties.getConsumerConcurrency());
      factory.setAutoStartup(true);
      factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);       
      return factory;
  }
  </code></pre>
  
  <p>Kafka retry listener:</p>
  
  <pre><code>@KafkaListener(topics = "${spring.kafka.retry.topic}", groupId = "${spring.kafka.consumer-group-id}", 
          containerFactory = "kafkaRetryListenerContainerFactory", id = "retry.id")
  public void retryMessage(ConsumerRecord&lt;String, String&gt; record, Acknowledgment acknowledgment) {
      Thread.sleep(900000);
      LOG.info(String.format("Consumed retry message -&gt; %s", record.toString()));
      acknowledgment.acknowledge();
  }
  </code></pre>
  
  <p>When I added Thread.sleep(), I'm getting continuous rebalancing error in the logs</p>
  
  <pre><code>Attempt to heartbeat failed since group is rebalancing
  </code></pre>
  
  <p>My spring kafka version is 2.3.4</p>
  
  <p>Below are the config values:</p>
  
  <p>max.poll.interval.ms = 1200000 (this is higher than thread.sleep)</p>
  
  <p>heartbeat.interval.ms = 3000</p>
  
  <p>session.timeout.ms = 10000</p>
  
  <p>I have tried ack.nack(900000); Still getting the rebalancing error</p>
  
  <p>Any help will be appreciated</p>