Apache NiFi - Asynchronously process messages after Kafka consumer

Question

Currently we are using Apache NiFi to consume messages via Kafka consumer. Output of kafka consumer is connected to DB processor which gets the messages in queue (from consumer) and runs the stored proc/processing on it. So the DB processor will be working on one message per from queue and I can set the DB processor to work in parallel for n threads, but primarily each thread can work on one message per queue.

I am looking to do something like below:

1. processor after consumer will just consume message (or take messages) from queue and say will wait for "batch" or total to 1000 messages.
2. As soon as it gets 1000 messages OR 60 secs passed and message count is < 1000, push to another processor which can be DB stored proc for business logic on group of those messages.
3. Mainly, I want above to be multithreaded i.e. if we get 3000 messages, the first processor will read them in 3 batches and push to DB processor (parallely).

So I want to know any such processor which can do point 2 above i.e. just read messages and push it to next based on batch/time rules?

Answer 1

If you can leverage NiFi's record processor's then using ConsumeKafkaRecord with a batch size of 1000 followed by PutDatabaseRecord will give you similar behavior to what you are describing.

If you may not always have enough messages available in the Kafka topic at the time of consuming, then adding MergeContent or MergeRecord in the middle would let you wait for a certain amount of time or number of messages.