Feature: Support automated offset management with external processing

[sehz]

To build high performance sink connector, connector needs to bulk commit records to sink after successfully commits to external data sink. This should be supported by automated batch offset.

It would be good have something like:

let consumer = let config = ConsumerConfigExtBuilder::default()
       .topic(TOPIC_NAME)
       .partition(PARTITION_NUM)
       .post_process(MyPostProcesser::new(), OffsetError:default)

MyPostProcessor will implement consumer trait that is sort of callback to control lifecycle:

pub trait ConsumerLifecycle {

       // there is default implementation, not need to override  
       async fn process_batch(&self, batch: BatchRecord) -> Result<()>

       async fn process_record(&self);
}

The Offset management will take care of offset in case of error.