title: Spring Kafka SeekToCurrentErrorHandler maxFailures doesn't work when concurrency level is less than partition number
body: <p>I am using spring kafka 2.2.7 and my consumer configuration code is following:</p>
  
  <pre><code>@Slf4j
  @Configuration
  @EnableKafka
  public class KafkaConfiguration {
  
    @Bean
    ConcurrentKafkaListenerContainerFactory&lt;String, Customer&gt; kafkaListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory&lt;String, Customer&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(consumerFactory());
      // less than number of partition, will do infinite retry
      factory.setConcurrency(1);
      SeekToCurrentErrorHandler errorHandler =
          new SeekToCurrentErrorHandler((record, exception) -&gt; {
            LOGGER.info("***in error handler data, {}", record);
          }, 1);
      factory.setErrorHandler(errorHandler);
      return factory;
    }
  
    @Bean
    public ConsumerFactory&lt;String, Customer&gt; consumerFactory() {
      return new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs());
    }
  
    @Bean
    public Map&lt;String, Object&gt; consumerConfigs() {
      Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
      props.put(ErrorHandlingDeserializer2.VALUE_DESERIALIZER_CLASS, KafkaAvroDeserializer.class);
  
      props.put("schema.registry.url", "http://127.0.0.1:8081");
      props.put("specific.avro.reader", "true");
  
      props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
      props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
      props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
      return props;
    }
  }
  
  
  @Component
  @Slf4j
  public class KafkaConsumerService {
  
    @KafkaListener(id = "demo-consumer-stream-group", topics = "kafka-demo-avro")
    public void process(ConsumerRecord&lt;String, Customer&gt; record) {
      LOGGER.info("Customer key: {} and value: {}", record.key(), record.value());
      LOGGER.info("topic: {}, partition: {}, offset: {}", record.topic(), record.partition(), record.offset());
      throw new RuntimeException("force to retry");
    }
  }
  </code></pre>
  
  <p>So, if I have an exception happening in my listener, the consumer will retry the failed message forever even I config the maxFailures in the configuration if the concurrency level I specified is less than the partition counts for my topic.</p>
  
  <p>It will only work if i send the message one by one at least a second interval. If I send the message as a batch, the behaviour wouldn't work. Unless I restart the consumer and it will work properly.</p>
  
  <p>Step to reproduce:
  1. create a topic with more than 1 partition, for example 3 or 6
  2. In Spring kafka config, specify concurrency level to 1
  3. For <code>SeekToCurrentErrorHandler</code>, specify <code>maxFailure</code> to a positive value for example 3
  4. Send dozen of messages to the topic</p>
  
  <p>You will see each failed message will do infinite retry instead of the <code>maxFailure</code> I specified. Also, I can see a lot of messages fall behind of the consumer lag.</p>
  
  <p>However, if you stop the listener and start the listener again, it will skip the failed messages properly.</p>