More than expected messages appearing in kafka dead letter topic

[TamimEhsan]

I am new to Apache kafka and trying to understand how the non blocking retry works.

I have two consumer in different groups. I tried to simulate errors manually

@RetryableTopic(attempts = "2" )// 1 retries
    @KafkaListener(topics = "email", groupId = "consumer1", concurrency = "1")
    public void consumeMessage(EmailDto message) throws Exception {
        System.out.println("Received message: " + message.getTo());
        if( message.getTo().equals("error1") ) throw new Exception("Error processing message");
        if( message.getTo().equals("error") ) throw new Exception("Error processing message"); 
    }

@RetryableTopic(attempts = "2" )// 1 retries
    @KafkaListener(topics = "email", groupId = "consumer2", concurrency = "1")
    public void consumeMessage(EmailDto message) throws Exception {
        System.out.println("Received message: " + message.getTo());
        if( message.getTo().equals("error2") ) throw new Exception("Error processing message");
        if( message.getTo().equals("error") ) throw new Exception("Error processing message"); 
    }

Now, when I send produce an event with value equals to error1 or error2 I get 1 message in email, 1 in email-retry and 1 in email-dlt as expected. But when I produce a event with value error then I get 1 message in email, 2 in email-retry (so far expected) but 4 messages in email-dlt.

Shouldn't the email-dlt topic also have 2 messages?

I think I should add some more information about the configuration too.
The kafka cluster is running in docker with the following compose

 kafka:
    image: confluentinc/cp-kafka:7.4.4
    depends_on:
      - zookeeper
    ports:
      - 29092:29092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_NUM_PARTITIONS: 2

The producer configs are

spring.kafka.producer.bootstrap-servers: localhost:29092
spring.kafka.producer.key-serializer: org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.properties.spring.json.type.mapping=person:com.example.producer.PersonDto,email:com.example.producer.EmailDto
spring.kafka.producer.acks: all
spring.kafka.producer.retries: 0
spring.kafka.producer.batch-size: 16384
spring.kafka.producer.properties.linger.ms: 1
spring.kafka.producer.properties.buffer.memory: 33554432

And the consumer config are

spring.kafka.bootstrap-servers=localhost:29092
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.producer.key-serializer: org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.properties.spring.json.json.type.mapping=person:com.example.consumer.PersonDto,email:com.consumer.common.EmailDto

spring.kafka.consumer.key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
spring.kafka.consumer.properties.spring.deserializer.value.delegate.class=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.type.mapping=person:com.example.consumer.PersonDto,email:com.example.consumer.EmailDto
spring.kafka.consumer.properties.spring.json.trusted.packages=*

The topics (email, email-retry, email-dlt) are generated automatically. The partition count and replication factor of all the topics are 1.