Spring kafka retry delay is shorter than configured

Question

I am using Spring Kafka as retry mechanism. I configured Kafka listener to retryable topic with following code:

@RetryableTopic(
            include= {SystemException.class},
            topicSuffixingStrategy= TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE,
            fixedDelayTopicStrategy = FixedDelayStrategy.MULTIPLE_TOPICS,
            retryTopicSuffix = "_retry",
            dltTopicSuffix = "_dlq",
            attempts = "3",
            backoff = @Backoff( delay = 3600000L, multiplier = 1 ),
            autoCreateTopics = "false")

As you can see, the delay of backoff is 3600000ms, which equals 1 hr. However, the retry actually happened about 36 seconds after an exception or error. Delay time is much more shorter than 1 hour.

I think "max.poll.interval.ms" make trigger the re-balance. So I have made "max.poll.interval.ms" longer than delay time as below:

max.poll.interval.ms = 3605000

That didn't resolve the issue. And I don't think I have rebalance issue because there is only one instance listening to the topic.

Do I need to set some properties for KafkaTemplate or ListenerConsumerFactory? Please help. Thank you so much in advance.

Answer 1

The poll interval has no bearing on this.

Turn on DEBUG logging and look for these: Caused by: org.springframework.kafka.listener.KafkaBackoffException: Partition 0 from topic so77360490-retry-0 is not ready for consumption, backing off for approx. 59483 millis. (in this case, the delay was 60000).

Then you will see DEBUG [so77360490-retry-0-0-C-1] Paused consumption from [so77360490-retry-0-0]

And it shouldn't resume until the time has expired.

With 3600000, I got Partition 0 from topic so77360490-retry-0 is not ready for consumption, backing off for approx. 3599491 millis.

EDIT

I let my test run for an hour, and an hour later, I see

Resumed consumption from [so77360490-retry-0-0]
Received: 1 records
Back off time: -42 ...
foo from so77360490-retry-0

(the last one from my listener:

    @KafkaListener(id = "so77360490", topics = "so77360490")
    @RetryableTopic(backoff = @Backoff(delay = 3600000))
    void listen(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        log.info(in + " from " + topic);
//      throw new RuntimeException();
    }