Implement timeout handling for when kafka records are not consumed at sufficient pace

**Description**
When kafka records are not being sent to the consumer at sufficient intervals, trigger flushing of the consumer's cache of intermediate results to HDFS. Implement configurable timeout parameter to configuration.