CODEWITHSUNDEEP
spring-bootspring-cloudkafka-consumer-apiapache-kafka-streams
nitin737
nitin737

Reputation: 69

Spring cloud kafka stream consumer retry mechanism

What is the meaning of the properties below and how do I use them?

spring.cloud.stream.bindings.atcommnity.consumer.maxAttempts=5
spring.cloud.stream.bindings.atcommnity.consumer.backOffInitialInterval=1000
spring.cloud.stream.bindings.atcommnity.consumer.backOffMaxInterval=2000000
spring.cloud.stream.bindings.atcommnity.consumer.backOffMultiplier=2.0
spring.cloud.stream.bindings.atcommnity.consumer.batch-mode=false

Upvotes: 0

Views: 820

Answers (2)

nitin737
nitin737

Reputation: 69

Above answer from Alex is the perfect explanation for this question. And here just to increase my knowledge and share some extra findings. I am providing the actual implementation from the code.

private BackOff createBackOff(
        final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {

    int maxAttempts = extendedConsumerProperties.getMaxAttempts();
    if (maxAttempts < 2) {
        return new FixedBackOff(0L, 0L);
    }
    int initialInterval = extendedConsumerProperties.getBackOffInitialInterval();
    double multiplier = extendedConsumerProperties.getBackOffMultiplier();
    int maxInterval = extendedConsumerProperties.getBackOffMaxInterval();
    ExponentialBackOff backOff = new ExponentialBackOff(initialInterval, multiplier);
    backOff.setMaxInterval(maxInterval);
    long maxElapsed = extendedConsumerProperties.getBackOffInitialInterval();
    double accum = maxElapsed;
    for (int i = 1; i < maxAttempts - 1; i++) {
        accum = accum * multiplier;
        if (accum > maxInterval) {
            accum = maxInterval;
        }
        maxElapsed += accum;
    }
    backOff.setMaxElapsedTime(maxElapsed);
    return backOff;
}

Class Name KafkaMessageChannelBinder

Upvotes: 0

Alex
Alex

Reputation: 5982

The backoff will start with backOffInitialInterval and then every next attempt will be multiplied by backOffMultiplier but will not exceed backOffMaxInterval.

currentInterval = Math.min(backOffInitialInterval * Math.pow(backOffMultiplier, retryNum), backOffMaxInterval)

In your case it will be:

1000ms -> 2000ms -> 4000ms -> 8000ms -> 16000ms

Upvotes: 2

Related Questions