I'm having a situation where 1 thread is consuming messages from a Kafka topic at a fast rate and putting them into a blocking queue that is then getting consumed in another thread writing bulk inserts into a mongo database collection. I have not seen a lot of answers as this is a common problem, my app is crashing because the message q gets so big and runs out of memory as the mongo db writer thread can't keep up with the message consumption rate.
What is the right way to configure a kafka consumer to pause message consumption for a period of time until the message q gets back to a reasonable size. Can I do a pause or something in the pool loop? I don't think so otherwise the consumer will be flagged as not online, could I close the Kafka consumer every time the message q gets to be too large and then reconnect when its back to a manageable size? I could but that does not seem like a clean solution what I am looking for is to say "hey kafka please pause the sending of messages to my active consumer until I tell you to resume" as this would allow me to pull messages at the speed I can insert them into my data store.
Please help!