title: Spring Kafka consumer not able to consume records
body: <p>We are using Spring Kafka to consume records in batches. We are sometimes facing an issue where the application starts and it doesn't consume any records even though there are enough unread messages. Instead we continuously see info logs saying. </p>
  
  <pre><code>[INFO]-[FetchSessionHandler:handleError:440] - [Consumer clientId=consumer-2, groupId=groupId] Error sending fetch request (sessionId=INVALID, epoch=INITIAL) to node 1027: org.apache.kafka.common.errors.DisconnectException. 
  </code></pre>
  
  <p>People are facing this issue and everyone says to ignore it, since it is just a info log. Even, we see after sometime the application starts picking up the records without doing anything. But, it is very unpredictable on how long it might take to start consuming records :( </p>
  
  <p>We didn't see this error when we were using Spring cloud stream. Not sure if we have missed any configuration in spring-kafka.</p>
  
  <p>Anyone faced this issue in past, please let us know if we are missing something. <strong>We have huge load in our topics and if there is a lot of lag, could this happen?</strong></p>
  
  <p>We are using <strong>Spring Kafka of 2.2.2.RELEASE</strong> 
  <strong>Spring boot 2.1.2.RELEASE</strong>
  <strong>Kafka 0.10.0.1</strong> (We understand it's very old, because of unavoidable reasons we are having to use this :()</p>
  
  <p>Here is our code:</p>
  
  <p>application.yml</p>
  
  <pre><code>li.topics: CUSTOM.TOPIC.JSON
      spring:
        application:
          name: DataPublisher
        kafka:
          listener:
            type: batch
            ack-mode: manual_immediate
          consumer:
            enable-auto-commit: false
            max-poll-records: 500
            fetch-min-size: 1
            fetch-max-wait: 1000
            group-id: group-dev-02
            key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
            value-deserializer:CustomResourceDeserialiser
            auto-offset-reset: earliest
  </code></pre>
  
  <p>Consumer:</p>
  
  <pre><code>public class CustomKafkaBatchConsumer {
  
  
    @KafkaListener(topics = "#{'${li.topics}'.split(',')}", id = "${spring.kafka.consumer.group-id}")
    public void receiveData(@Payload List&lt;CustomResource&gt; customResources,
        Acknowledgment acknowledgment,
        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List&lt;Integer&gt; partitions,
        @Header(KafkaHeaders.OFFSET) List&lt;Long&gt; offsets) {
  }
  }
  </code></pre>
  
  <p>Deserialiser:</p>
  
  <pre><code>public class CustomResourceDeserialiser implements Deserializer&lt;CustomResource&gt; {
  
    @Override
    public void configure(Map&lt;String, ?&gt; configs, boolean isKey) {
    }
  
    @Override
    public CustomResource deserialize(String topic, byte[] data) {
      if (data != null) {
        try {
          ObjectMapper objectMapper = ObjectMapperFactory.getInstance();
          return objectMapper.readValue(data, CustomResource.class);
        } catch (IOException e) {
          log.error("Failed to deserialise with {}",e.getMessage());
        }
      }
      return null;
    }
  
    @Override
    public void close() {
  
    }
  }
  </code></pre>