Can Spring-Kafka support rate limit

[xzhan]

@garyrussell @artembilan

We have one interesting requirement on Consume message with rate limit

On the peak time, we have huge message generate on Kafka server, and if we keep default consume config like

spring.kafka.consumer.max-poll-records=500
spring.kafka.consumer.fetch-max-wait=500

we do some change like below, reducer poll records and change the fetch-max-wait time from 500 to 1000

spring.kafka.consumer.max-poll-records=100
spring.kafka.consumer.fetch-max-wait=1000

if it take effective ,it means it will do some rate limit , am i right? do we have another way to do it with Spring-Kafka

more than that , on peak time to reduce consumer server high cpu we want it consume slowly then back to normal traffic we can recover original consume speed?

I know we can add this topic partition ,and add more consumer node to handle it. but not sure whether we have more effective program way to do dynamic consume

[artembilan]

Those props are static for the consumer. If we can change them at runtime, we have to restart a listener container to apply new configs.
See ConsumerFactory.updateConfigs(Map<String, Object> updates).
According to this article it won't help too much though: https://www.linkedin.com/pulse/throttling-kafka-consumption-dispelling-myths-ganesh-ghag/.

As long as you have enough data in the topic log, those props don't matter: the consumer is going to fetch according config, call listener and come back to other records in the topic.

You may consider to use an idleBetweenPolls property though: https://docs.spring.io/spring-kafka/reference/html/#idleBetweenPolls.
But again: you have to restart container to apply its changes at runtime.

[xzhan]

Thx @artembilan our service have try idleBetweenPolls property , as you mentioned it is static config. but that means we need always slow down our consume speed as we set Idle.

Also we real do some setting as you share the wiki , I vote we can do more tuning on the fetch.min.bytes we did not turn before

[xzhan]

But I still have another idea want to discuss with you. On normal status not peak time our max.poll.records not always reach the max value , eg we set value = 500 . But at the peak time it will nearly reach 500, so we can think that time is Peak time , if build ratelimit concept can we do delay ack to Kafka to slow down the whole transcation.

-cc @artembilan and @garyrussell what's you point

[xzhan]

also @artembilan i checked the function
/**
* Update the consumer configuration map; useful for situations such as
* credential rotation.
* @param updates the configuration properties to update.
* @SInCE 2.7
*/
default void updateConfigs(Map<String, Object> updates) {
}

do we have example while we use @KafkaListener condition , then we dynamic refresh consumerConfig without restart as this case defined useful for situations such as credential rotation

[artembilan]

It cannot be done without restarting a listener container.
Those properties are set into a KafkaConsumer which is not mutable, so to apply a fresh config we have to close current consumers a restart their container to read new configs and start new consumers.
This is pretty simple and lightweight operation from Spring application perspective.
Right, such a restart causes a rebalance on Kafka broker, but there is indeed no other way to apply fresh config into Kafka client by itself.

The ack has nothing to do with fetch. It even fully can be skipped from the logic.

The mentioned idleBetweenPolls is your best bet to "slow down the whole transcation".

You may consider to add a rate limit on the listener (client) side.
See Resilience4j RateLimiter

This way the call to your function is going to be back-pressured and that, in turn, will push a Kafka consumer for its poll() as well.
Of course this must happen within pollTimeout interval.
Or you can react to such a back-pressure and pause() your consumer.
See more in docs: https://docs.spring.io/spring-kafka/reference/html/#pause-resume