Today we only update the consumer group offset for a processor every 5 minutes if it does not have a window and every 1 hour if it has a window. This is a poor user experience has many customers have infrastructure and integrations that monitor the Kafka consumer group offsets movement and calculate total lag for alerts. We should change this so the offsets are updated in near-real time to support the expected behavior customers have around kafka consumer group offsets. 

To configure this, we use the same Kafka property - enable.auto.commit and the default is true

$source: { config:

}

When auto.commit is true, we update the consumer group as we read messages. When auto.comit is false, we update the consumer group offsets only after the corresponding offsets are committed in a checkpoint and flushed to S3 (this is already implemented).