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);
			}
		}
	}
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.

@param internalOffsets The internal offsets (representing last processed records) for the partitions to commit.
@throws Exception The method forwards exceptions.