public void prepareAndCommitOffsets ( Map < KafkaTopicPartition , Long > internalOffsets ) throws Exception { for ( Map . Entry < KafkaTopicPartition , Long > entry : internalOffsets . entrySet ( ) ) { KafkaTopicPartition tp = entry . getKey ( ) ; Long lastProcessedOffset = entry . getValue ( ) ; if ( lastProcessedOffset != null && lastProcessedOffset >= 0 ) { setOffsetInZooKeeper ( curatorClient , groupId , tp . getTopic ( ) , tp . getPartition ( ) , lastProcessedOffset + 1 ) ; } } }
Commits offsets for Kafka partitions to ZooKeeper . The given offsets to this method should be the offsets of the last processed records ; this method will take care of incrementing the offsets by 1 before committing them so that the committed offsets to Zookeeper represent the next record to process .