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