Default-retryable setting is ignored

[nataliyachabini]

Describe the issue

Even if default-retryable is set to false, the messages are retried. It is the same for retryable-exceptions.

There also seems to be a connection with the transactions, as once the transaction-id-prefix is empty, it works as intended.

To Reproduce

SpringBootApplication

@Configuration
@Slf4j
public class ConsumerConfig {
    @Bean
    public Consumer<String> consumeMessage() {
        return s -> {
            log.info("Consuming {}", s);
            throw new IllegalArgumentException(s);
        };
    }
}

application.xml

  cloud:
    function:
      definition: consumeMessage
    stream:
      kafka:
        binder:
          transaction:
            transaction-id-prefix: transaction-
          required-acks: all
          configuration:
            key.serializer: org.apache.kafka.common.serialization.StringSerializer
            key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
        bindings:
          consumeMessage-in-0:
            consumer:
              enable-dlq: true
      bindings:
        consumeMessage-in-0:
          group: my-group
          destination: my-topic
          consumer:
            default-retryable: false
            max-attempts: 5
            back-off-initial-interval: 100
            retryable-exceptions:
              java.lang.UnsupportedOperationException: true
              java.lang.IllegalArgumentException: false

There is a GitHub project with the minimal setup to reproduce the issue:
https://github.com/DidierLoiseau/kafka-transactions-and-retries/tree/main

When we submit a message on my-topic, it will retry it 5 times despite the default-retryable: false and java.lang.IllegalArgumentException: false.

We have asked a question on StackOverflow but (so far) have not gotten any answers: https://stackoverflow.com/questions/79309828/how-to-configure-retryable-exceptions-for-consumers-when-kafka-transactions-are

Version of the framework
Spring Boot 3.4.0 and Spring Cloud 2024.0.0

Additional context
Diving into the Spring Cloud Stream code, we found that KafkaMessageChannelBinder will set a RetryTemplate configured by buildRetryTemplate(properties) if there is no TransactionManager, but if there is one, it will configure an AfterRollbackProcessor instead, passing it only a BackOff without using the retryable exceptions configuration.

[olegz]

Thank you for reporting it. Looking. . .

[olegz]

I am still looking and trying to get clarification from the Kafka team, but it appears that introduction of transaction manager is mutually exclusive to binder retry settings.
Basically if you remove the following

transaction:
            transaction-id-prefix: transaction-
          required-acks: all

. . . then everything works

[nataliyachabini]

Thank you @olegz, this is indeed what we also have observed, that they cannot be used simultaneously.

[sobychacko]

@nataliyachabini Normal binding retry mechanisms are disabled when using transactions. See this note in the ref docs:

Normal binder retries (and dead lettering) are not supported with transactions because the retries will run in the original transaction, which may be rolled back and any published records will be rolled back too. When retries are enabled (the common property maxAttempts is greater than zero) the retry properties are used to configure a DefaultAfterRollbackProcessor to enable retries at the container level. Similarly, instead of publishing dead-letter records within the transaction, this functionality is moved to the listener container, again via the DefaultAfterRollbackProcessor which runs after the main transaction has rolled back.

https://docs.spring.io/spring-cloud-stream/reference/kafka/kafka-binder/transactional.html

When transactions are enabled in Spring Cloud Stream Kafka binder, retries are handled differently:

Binder-level retry settings (default-retryable, retryable-exceptions) are bypassed. Instead, retry behavior is controlled by the DefaultAfterRollbackProcessor at the container level.
To disable retries, you must configure the AfterRollbackProcessor directly using the ListenerContainerCustomizer:

public ListenerContainerCustomizer<AbstractMessageListenerContainer<byte[], byte[]>> customizer() {
    return (container, destination, group) -> container.setAfterRollbackProcessor(
        new DefaultAfterRollbackProcessor<byte[], byte[]>(
            (record, exception) -> log.error("Failed to process: {}", exception.getMessage()),
            new FixedBackOff(0L, 0)  // Disables retries
        )
    );
}

Here is another example matching your retry settings.

@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer<byte[], byte[]>> customizer() {
    return (container, destination, group) -> container.setAfterRollbackProcessor(
        new DefaultAfterRollbackProcessor<byte[], byte[]>(
            (record, exception) -> {
                if (exception instanceof IllegalArgumentException) {
                    log.error("Not retrying IllegalArgumentException: {}", exception.getMessage());
                } 
                else {
                    throw new RuntimeException(exception); // Allow retry for other exceptions
                }
            },
            new ExponentialBackOff(100, 5)) // Matches your original backoff settings
    );
}

Here is a blog that might be related: https://spring.io/blog/2023/10/11/transactional-rollback-strategies-with-spring-cloud-stream-and-apache-kafka

[olegz]

@nataliyachabini So, given the explanation, does it answer your question?

[nataliyachabini]

Yes, thank you for your help @olegz @sobychacko