spring-kafka - When there are multiple failed records from different partitions seekToCurrentErrorHandler behaves random and consumer leaves broker

Question

I am looking into recovering error records using a custom recoverer with seekToCurrentErrorHandler. We are using spring-kafka version 2.2.6-RELEASE (with spring-boot) in our project. Saw that (and faced the issue) in case of a single consumer (and concurrency 1) with multiple failed records from different partitions it ends up in infinite loop ( https://github.com/spring-projects/spring-kafka/issues/1237) As its fixed in later versions, have switched to 2.3.3-RELEASE, and ended up facing another issue described below (tried version 2.5.4-RELEASE as well, ending up in same issue).

So with spring-kafka 2.3.3-RELEASE or 2.5.4-RELEASE below is the issue we are facing.

When a single kafka consumer is brought up with 6 records to be consumed from a topic (with 3 partitions, 2 records on each partition) and when all of them result in failure (@KafkaListener annotated method throwing a runtimeException) listener (i.e. @KafkaListener annotated method) gets invoked for 2 or 3 records (random), does retries as expected (5 retires + 1 original failure) and recoverer gets invoked for those records and for 3rd or 4th record about 3 times retry is attempted and after about 5mins (which is default max.poll.interval.ms ) consumer disconnects from broker (doesn't process rest of the 3 records) . Also when 3 records are processed, only 2 are committed (i.e. next time when app is started, listener is again invoked with already processed 3rd record) Below is the configuration. Could be replicated with JUNIT as well. Please let me know if I am missing something here.!! Can we not use only seekToCurrentErrorHandler with recoverer in case of multiple failed records from different partitions? or we have to use stateful retry with RecoveryCallback only ?(tried that as well and saw same issue) ?

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) throws IOException{
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.getContainerProperties().setSyncCommits(true);
        factory.getContainerProperties().setSyncCommitTimeout(Duration.ofSeconds(2000));
        SeekToCurrentErrorHandler seekToCurrentErrorHandler = new SeekToCurrentErrorHandler(processingFailedErrorRecoverer(),new FixedBackOff(0L,5L));
        seekToCurrentErrorHandler.setCommitRecovered(true);
        factory.setErrorHandler(seekToCurrentErrorHandler);
        return factory;
    }

With retryTemplate and recoverycallback

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) throws IOException{
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.getContainerProperties().setSyncCommits(true);
        factory.setRetryTemplate(new RetryTemplate()); // 3 retries by default
        factory.setStatefulRetry(true);
        factory.setRecoveryCallback(context -> {
            processingFailedErrorRecoverer().accept((ConsumerRecord<?, ?>) context.getAttribute("record"),
                    (Exception) context.getLastThrowable());
            return null;
        });

        factory.getContainerProperties().setSyncCommitTimeout(Duration.ofSeconds(2000));
        SeekToCurrentErrorHandler seekToCurrentErrorHandler = new SeekToCurrentErrorHandler(new FixedBackOff(0L,3L));
        seekToCurrentErrorHandler.setCommitRecovered(true);
        factory.setErrorHandler(seekToCurrentErrorHandler);
        return factory;
    }

Log statements when consumer disconnects.

2.3.3-RELEASE -> 2020-08-10 10:51:23.881 23 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=xyz-consumer-group] Member consumer-1-1a0978c4-9ae6-45b9-8d9d-f3ddee081df9 sending LeaveGroup request to coordinator (id: 2147482644 rack: null) 2020-08-10 10:51:23.882 23 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO org.apache.kafka.clients.consumer.KafkaConsumer - [Consumer clientId=consumer-1, groupId=xyz-consumer-group] Unsubscribed all topics or patterns and assigned partitions

2.5.4-RELEASE -> 2020-08-10 14:34:20.902 36 [kafka-coordinator-heartbeat-thread | xyzconsumer-group] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-xyz-consumer-group-1, groupId=xyz-consumer-group] Member consumer-xyz-consumer-group-1-8324f4e3-4ec3-4b34-b6af-c4ff01a0aa01 sending LeaveGroup request to coordinator (id: 2147482644 rack: null) due to consumer poll timeout has expired. This means the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.

src/main/java

package com.orgname.gtb.cmng.kafka;
/**
 * @param <V> Original message type.
 * @param <T> Message type to be published.
 */
@Slf4j
public abstract class AbstractErrorRecoverer<V,T> implements BiConsumer<ConsumerRecord<?, ?>, Exception> {
    private static final String LOGGER_NAME="ERRORHANDLER";
    private static final Logger LOGGER = LoggerFactory.getLogger(LOGGER_NAME);       
    private final KafkaTemplate<String, T> kafkaTemplate; 
    private final KafkaTemplate<String, byte[]> deserializationErrorRecoveryKafkaTemplate;

    protected AbstractErrorRecoverer(KafkaTemplate<String, T> kafkaTemplate,KafkaTemplate<String, byte[]> deserializationErrorRecoveryKafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
        this.deserializationErrorRecoveryKafkaTemplate=deserializationErrorRecoveryKafkaTemplate;
        log.info("Recoverer initialized with alertDispatcher and kafkaTemplate.");
    }
    

    @SuppressWarnings("unchecked")
    @Override
    public void accept(ConsumerRecord<?, ?> consumerRecord, Exception e) {
        V original = (V) consumerRecord.value();
        // TODO Do other common things, like alerting etc. 
        List<Header> headers = this.enhanceHeaders(consumerRecord, e);
        
        DeserializationException deserEx = ListenerUtils.getExceptionFromHeader(consumerRecord,
                ErrorHandlingDeserializer.VALUE_DESERIALIZER_EXCEPTION_HEADER,  new LogAccessor(LOGGER_NAME));
        
        if(deserEx!=null){  
            ProducerRecord<String, byte[]> deserilizationErrorRecord = new ProducerRecord<>(getDeserializationErrorRecoveryTopic(), consumerRecord.partition(),
                    (String) consumerRecord.key(), deserEx.getData(), headers);
            if (deserializationErrorRecoveryKafkaTemplate.isTransactional() && !deserializationErrorRecoveryKafkaTemplate.inTransaction()) {
                deserializationErrorRecoveryKafkaTemplate.executeInTransaction(kafkaOperations -> {
                    this.publishDeserializationError(deserilizationErrorRecord, kafkaOperations);
                    return null;
                });
            } else {
                publishDeserializationError(deserilizationErrorRecord, deserializationErrorRecoveryKafkaTemplate);
            }
        }
        else {
            T objectToPublish=messageToPublish(consumerRecord,e.getCause());
            ProducerRecord<String, T> pr = new ProducerRecord<>(getErrorTopic(), consumerRecord.partition(),
                    (String) consumerRecord.key(), objectToPublish, headers);

            if (kafkaTemplate.isTransactional() && !kafkaTemplate.inTransaction()) {
                kafkaTemplate.executeInTransaction(kafkaOperations -> {
                    this.publish(pr, kafkaOperations);
                    return null;
                });
            } else {
                publish(pr, kafkaTemplate);
            }
        }
    }

    private void publish(ProducerRecord<String, T> record, KafkaOperations<String, T> ops) {
        try {   
            ops.send(record).addCallback(stringTSendResult -> {
                log.debug("Successfully published message to dead letter topic");
            }, ex -> {
                log.error("error publishing to ERROR-Topic", ex);
            });
        } catch (Exception e) {
            log.error("Error publishing to error-topic.", e);
        }
    }
    
    private void publishDeserializationError(ProducerRecord<String, byte[]> record, KafkaOperations<String, byte[]> ops) {
        try {
            System.out.println("before pub to recovery topic");
            ops.send(record).addCallback(stringTSendResult -> {
                log.debug("Successfully published message to deserialization recovery topic.");
            }, ex -> {
                log.error("error publishing to deserialization recovery topic.", ex);
            });
        } catch (Exception e) {
            log.error("Error publishing to deserialization recovery topic.", e);
        }
    }
    
    
    private List<Header> enhanceHeaders(ConsumerRecord<?, ?> record, Exception exception) {
        List<Header> headers = new ArrayList<>();
        headers.add(new RecordHeader(KafkaHeaders.DLT_ORIGINAL_TOPIC, record.topic().getBytes(StandardCharsets.UTF_8)));
        headers.add(new RecordHeader(KafkaHeaders.DLT_ORIGINAL_PARTITION, ByteBuffer.allocate(4).putInt(record.partition()).array()));
        headers.add(new RecordHeader(KafkaHeaders.DLT_ORIGINAL_OFFSET, ByteBuffer.allocate(8).putLong(record.offset()).array()));
        headers.add(new RecordHeader(KafkaHeaders.DLT_ORIGINAL_TIMESTAMP, ByteBuffer.allocate(8).putLong(record.timestamp()).array()));
        headers.add(new RecordHeader(KafkaHeaders.DLT_ORIGINAL_TIMESTAMP_TYPE, record.timestampType().toString().getBytes(StandardCharsets.UTF_8)));
        headers.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_FQCN, exception.getClass().getName().getBytes(StandardCharsets.UTF_8)));
        headers.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_MESSAGE, exception.getMessage().getBytes(StandardCharsets.UTF_8)));
        headers.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_STACKTRACE, this.getStackTraceAsString(exception).getBytes(StandardCharsets.UTF_8)));
        Header valDeserExceptionheader  =record.headers().lastHeader(ErrorHandlingDeserializer.VALUE_DESERIALIZER_EXCEPTION_HEADER);
        if (valDeserExceptionheader != null) {
            headers.add(valDeserExceptionheader);
        }
        return headers;
    }

    private String getStackTraceAsString(Throwable cause) {
        StringWriter stringWriter = new StringWriter();
        PrintWriter printWriter = new PrintWriter(stringWriter, true);
        cause.printStackTrace(printWriter);
        return stringWriter.getBuffer().toString();
    }

    /**
     * @return  The error topic to which the notification should be sent.
     */
    protected abstract String getErrorTopic();
    
    /**
     * 
     * @return The error topic to which deserialization error should be sent.
     */
    protected abstract String getDeserializationErrorRecoveryTopic();
    
    /**
     * This method receives the original consumer record and throwable that was thrown by the listener 
     * Override this method to publish a different message (e.g. an enriched message to errorTopic).
     * By default the original message is returned which is published.
     * @param originalConsumerRecord The original consumer record. Same as that received by listener
     * @param t Throwable thrown by listner.
     * @return The expected message to be published.
     */
    protected T messageToPublish(ConsumerRecord<?, ?> originalConsumerRecord,Throwable t){
        return (T)originalConsumerRecord.value();
    }

}

src/test/java

package com.orgname.gtb.cmng.config;
@EnableKafka
@Configuration
@Slf4j
public class IntegrationTestConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    // start of config for kafkatemplate that publishes a message 
    @Bean
    public Map<String, Object> producerProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        
        return props;
    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory(producerProps(), new StringSerializer(), new StringSerializer());
    }


    @Bean 
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
    //end of config for kafkatemplate that publishes a message
    
    // start of config for kafkatemplate that recovers deserialiazation error
    @Bean
    public Map<String, Object> deserializationErrorProducerProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        return props;
    }

    @Bean
    public ProducerFactory<String, byte[]> deserializationErrorProducerFactory() {
        return new DefaultKafkaProducerFactory(deserializationErrorProducerProps());
    }


    @Bean 
    public KafkaTemplate<String, byte[]> deserializationErrorRecoveryKafkaTemplate() {
        return new KafkaTemplate<>(deserializationErrorProducerFactory());
    }
   // end of config for kafkatemplate that recovers deserialiazation error
    
    // config for kafkatemplate that publishes to deadlettertopic.
    @Bean
    public KafkaTemplate<String, String> deadLetterKafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
    
    // consumers config
    @Bean
    public Map<String, Object> getConsumerProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 15000);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        return props;
    }

    @Bean
    DefaultKafkaConsumerFactory consumerFactory() {
        return new DefaultKafkaConsumerFactory(
                getConsumerProps(),
                new StringDeserializer(),
                new StringDeserializer()
        );
    }

    // config for the error handler and its publisher to the dead letter topic

    @Bean   // the error recoverer
    public StringErrorRecovererImplementation processingFailedErrorRecoverer() {
        return new StringErrorRecovererImplementation(deadLetterKafkaTemplate(),deserializationErrorRecoveryKafkaTemplate());
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory){
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.getContainerProperties().setSyncCommits(true);
        factory.getContainerProperties().setSyncCommitTimeout(Duration.ofSeconds(2000));
        SeekToCurrentErrorHandler seekToCurrentErrorHandler = new SeekToCurrentErrorHandler(processingFailedErrorRecoverer(),new FixedBackOff(0L,5L));
        seekToCurrentErrorHandler.setCommitRecovered(true);
        factory.setErrorHandler(seekToCurrentErrorHandler);
        return factory;
    }
    
    // config for the listener on the happy topic
    @Bean
    @Primary
    public KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry() {
        KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry =
                new KafkaListenerEndpointRegistry();
        return kafkaListenerEndpointRegistry;
    }

    // the listener
    @Bean
    public IntegrationTestMessageListener simpleStringMessageListener() {
        return new IntegrationTestMessageListener(kafkaListenerEndpointRegistry());
    }

src/test/java

 package com.orgname.gtb.cmng.kafka.integrationtest;
    @RunWith(SpringRunner.class)
    @TestPropertySource(properties = {"spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}"})
    @EmbeddedKafka(partitions = 3, topics = {"${topics.happy}", "${topics.deadLetter}"})
    @SpringBootTest(classes = {IntegrationTestConfig.class})
    public class ErrorRecovererIntegrationTest {
    
         private static final String BAD_MESSAGE = "Poison message";
    
        @Value("${topics.happy}")
        private String happyTopic;
    
        @Value("${topics.deadLetter}")
        private String deadLetterTopic;
    
        @Autowired
        private EmbeddedKafkaBroker embeddedKafka;
    
        @Autowired
        private ConsumerFactory<String, String> consumerFactory; // will use the deadLetterConsumer factory in the TestKafkaConfig
    
    
        @Autowired
        protected KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
    
        @Autowired
        private IntegrationTestMessageListener listener;
    
        private Consumer<String, String> deadLetterConsumer;
    
        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;
    
        @Before
        public void setUp() {
            embeddedKafka.brokerProperty("controlled.shutdown.enable", true);
    
            for (MessageListenerContainer messageListenerContainer : kafkaListenerEndpointRegistry.getListenerContainers()) {
                log.debug("Listener container props:\n{}",messageListenerContainer.getContainerProperties().toString());
                ContainerTestUtils.waitForAssignment(messageListenerContainer, embeddedKafka.getPartitionsPerTopic());
            }
            deadLetterConsumer = consumerFactory.createConsumer();
            deadLetterConsumer.subscribe(Collections.singleton(deadLetterTopic));
            deadLetterConsumer.poll(Duration.ofMillis(0));
        }
    
        @After
        public void tearDown() {
            listener.clear();
        }
    
        @Test
        @DirtiesContext
        public void given_bad_message_should_publish_to_dead_letter_topic() throws Exception {
            IntStream.range(0, 6).forEach(i -> kafkaTemplate.send(happyTopic, i % 3,i+"", BAD_MESSAGE));
            Thread.sleep(5000);
            ConsumerRecords<String, String> consumerRecords= KafkaTestUtils.getRecords(deadLetterConsumer);
            assertEquals(6,consumerRecords.count());
        }

src/test/java

package com.db.orgname.cmng.kafka.integrationtest;
/**
 * This listener will listen for "poison messages" and throw a runtime exception so the exception handling can be done.
 */
@Service
@Slf4j
public class IntegrationTestMessageListener {

    @Getter
    private final KafkaListenerEndpointRegistry registry;

    @Getter
    private Map<String,String> messages = new HashMap<>();

    public void clear() {
        messages.clear();
    }

    @Autowired
    public IntegrationTestMessageListener(KafkaListenerEndpointRegistry registry) {
        log.debug("Created simple listener");
        this.registry = registry;
    }

    @KafkaListener(topics = "${topics.happy}")
    public void listen(@Payload String value, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key) {
        log.info("Simple listener received message --  key: {}, value: {}", key, value);
        if (value.toLowerCase().startsWith("poison")) {
            throw new RuntimeException("failed");
        } else {
            messages.put(key, value);
        }

    }

src/test/java

package com.orgname.gtb.cmng.kafka.integrationtest;
@Getter
@Service
public class StringErrorRecovererImplementation extends AbstractErrorRecoverer<String,String> {

    public StringErrorRecovererImplementation(KafkaTemplate<String, String> kafkaTemplate,KafkaTemplate<String, byte[]> deserializationErrorRecoveryKafkaTemplate) {
        super(kafkaTemplate,deserializationErrorRecoveryKafkaTemplate);
    }

    @Override
    protected String getErrorTopic() {
        return "T-ERROR-TOPIC";
    }

    @Override
    protected String messageToPublish(ConsumerRecord<?, ?> orginal, Throwable t) {
        String originalString=(String)orginal.value();
        return originalString + t.getMessage();
    }

    @Override
    protected String getDeserializationErrorRecoveryTopic() {
        return "T-DESERIALIZATION-ERROR-TOPIC";
    }

src/test/resources application.yml

topics:
  happy: T-HAPPY-TOPIC
  deadLetter: T-ERROR-TOPIC
  deserializationError: T-DESERIALIZATION-ERROR-TOPIC
spring:
  kafka:
    consumer:
      enable-auto-commit: false
      auto-offset-reset: latest
    producer:
      acks: all

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.orgname.gtb.cmng</groupId>
    <artifactId>nextgen-commons-error-handler</artifactId>
    <version>0.1.1-SNAPSHOT</version>
    <name>nextgen-commons-error-handler</name>
    <description>nextgen commons error handler</description> <!--fixme: Add proper description-->

    <properties>
        <java.version>1.8</java.version>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <sonar.language>java</sonar.language>
        <lombok.version>1.18.8</lombok.version>

        <!--Test Dependencies-->
        <confluent.version>5.4.0</confluent.version>
        <mockito-core.version>2.9.0</mockito-core.version>
        <mockito-all.version>1.9.5</mockito-all.version>
        <junit.version>4.13</junit.version>
        <assertj-core.version>3.13.2</assertj-core.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <scope>provided</scope>
            <version>${lombok.version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.5.4.RELEASE</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro</artifactId>
            <version>1.9.1</version>
            <exclusions>
                <exclusion>
                    <groupId>com.fasterxml.jackson.core</groupId>
                    <artifactId>jackson-core</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.fasterxml.jackson.core</groupId>
                    <artifactId>jackson-databind</artifactId>
                </exclusion>
            </exclusions>             
        </dependency>
        <!--Test Dependencies-->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <version>2.5.4.RELEASE</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>kafka-avro-serializer</artifactId>
            <version>${confluent.version}</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>kafka-schema-registry-client</artifactId>
            <version>${confluent.version}</version>
            <scope>test</scope>
            <exclusions>
              <exclusion>
                <groupId>com.fasterxml.jackson.core</groupId>
                <artifactId>jackson-databind</artifactId>
              </exclusion>
              <exclusion>
                    <groupId>com.fasterxml.jackson.core</groupId>
                    <artifactId>jackson-annotations</artifactId>
                </exclusion>
            </exclusions> 
        </dependency>
        <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>common-config</artifactId>
            <version>${confluent.version}</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>common-utils</artifactId>
            <version>${confluent.version}</version>
            <scope>test</scope>
        </dependency>
        <!-- Test dependencies -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <version>2.3.2.RELEASE</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>${junit.version}</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.assertj</groupId>
            <artifactId>assertj-core</artifactId>
            <version>${assertj-core.version}</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.jacoco</groupId>
                <artifactId>jacoco-maven-plugin</artifactId>
                <version>0.8.3</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>prepare-agent</goal>
                        </goals>
                    </execution>
                    <execution>
                        <id>report</id>
                        <phase>prepare-package</phase>
                        <goals>
                            <goal>report</goal>
                        </goals>
                        <!--TODO-changeme: Change the exclusions based on individual project requirements-->
                        <configuration>
                            <excludes>
                                <exclude>**/entities/*.class</exclude>
                                <exclude>**/avro/*.class</exclude>
                            </excludes>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.codehaus.mojo</groupId>
                <artifactId>sonar-maven-plugin</artifactId>
                <version>3.7.0.1746</version>
            </plugin>
        </plugins>
    </build>

</project>

Answer 1

Please provide a project that exhibits this behavior; I cannot reproduce it; everything works as expected with this application:

@SpringBootApplication
public class So63349172Application {

    public static void main(String[] args) {
        SpringApplication.run(So63349172Application.class, args);
    }

    @KafkaListener(id = "so63349172", topics = "so63349172")
    public void listen(String in) {
        System.out.println(in);
        throw new RuntimeException("test");
    }

    @Bean
    ErrorHandler eh() {
        return new SeekToCurrentErrorHandler(
                (rec, ex) -> System.out.println("Recovered " + ListenerUtils.recordToString(rec, true)),
                        new FixedBackOff(0, 2)) {

            @Override
            public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records,
                    Consumer<?, ?> consumer, MessageListenerContainer container) {

                System.out.println("Failed " + ListenerUtils.recordToString(records.get(0), true));
                super.handle(thrownException, records, consumer, container);
            }

        };
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("so63349172").partitions(3).replicas(1).build();
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, String> template) {
        return args -> {
            IntStream.range(0, 6).forEach(i -> template.send("so63349172", i % 3, null, "foo"));
        };
    }

}

I don't see anything in your configuration that would cause a rebalance.

Here are the results of my test (3 broker cluster)...

$ egrep '^(Failed|Recovered)' ../tmp/gg
Failed so63349172-1@0
Failed so63349172-2@0
Failed so63349172-0@0
Failed so63349172-1@0
Failed so63349172-2@0
Failed so63349172-0@0
Failed so63349172-1@0
Recovered so63349172-1@0
Failed so63349172-2@0
Recovered so63349172-2@0
Failed so63349172-0@0
Recovered so63349172-0@0
Failed so63349172-1@1
Failed so63349172-2@1
Failed so63349172-0@1
Failed so63349172-1@1
Failed so63349172-2@1
Failed so63349172-0@1
Failed so63349172-1@1
Recovered so63349172-1@1
Failed so63349172-2@1
Recovered so63349172-2@1
Failed so63349172-0@1
Recovered so63349172-0@1