How to limit Message consumption rate of Kafka Consumer in SpringBoot? (Kafka Stream)

Misubushi 0
Viewed 71

I want to limit my Kafka Consumer message consumption rate to 1 Message per 10 seconds .I'm using kafka streams in Spring boot .

Following is the property I tried to Make this work but it didn't worked out s expected(Consumed many messages at once).

config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokersUrl);
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
         config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
         //
         config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,1);
         config.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 10000);

is there any way to Manually ACK(Manual offsetCommits) in KafkaStreams? which will be usefull to control the msg consumption rate .

Please note that i'm using Kstreams(KafkaStreams) Any help is really appreciated . :)

apache-kafka-streams kafka-consumer-api apache-kafka spring-boot
2 Answers

I think you misunderstand what MAX_POLL_INTERVAL_MS_CONFIG actually does.

That is the max allowed time for the client to read an event.

From docs

controls the maximum time between poll invocations before the consumer will proactively leave the group (5 minutes by default). The value of the configuration request.timeout.ms (default to 30 seconds) must always be smaller than max.poll.interval.ms(default to 5 minutes), since that is the maximum time that a JoinGroup request can block on the server while the consumer is rebalance

"maximum time" not saying any "delay" between poll invocations.

Kafka Streams will constantly poll; you cannot easily pause/start it and delay record polling.

To read an event every 10 seconds without losing consumers in the group due to lost heartbeats, then you should use Consumer API, with pause() method, call Thread.sleep(Duration.ofSeconds(10)), then resume() + poll() while setting max.poll.records=1

OneCricketeer0

Finally ,I achieved the desired message consuming limit using Thread.sleep(). Since , there is no way to control the message consumption rate using kafka config properties itself . I had to use my application code to control the rate of consumption . Example: if I want control the record consumption rate say 4 msg per 10 seconds . Then i will just consumer 4 msg (will keep a count parallely) once 4 records are consumer then i will make the thread sleep for 10 seconds and will repeat the same process over again . I know it's not a good solution but there was no other way. thank you OneCricketeer

Misubushi0
Related