code: @Override
  	protected void doCommitInternalOffsetsToKafka(
  			Map<KafkaTopicPartition, Long> offsets,
  			@Nonnull KafkaCommitCallback commitCallback) throws Exception {
  
  		ZookeeperOffsetHandler zkHandler = this.zookeeperOffsetHandler;
  		if (zkHandler != null) {
  			try {
  				// the ZK handler takes care of incrementing the offsets by 1 before committing
  				zkHandler.prepareAndCommitOffsets(offsets);
  				commitCallback.onSuccess();
  			}
  			catch (Exception e) {
  				if (running) {
  					commitCallback.onException(e);
  					throw e;
  				} else {
  					return;
  				}
  			}
  		}
  
  		// Set committed offsets in topic partition state
  		for (KafkaTopicPartitionState<TopicAndPartition> partition : subscribedPartitionStates()) {
  			Long offset = offsets.get(partition.getKafkaTopicPartition());
  			if (offset != null) {
  				partition.setCommittedOffset(offset);
  			}
  		}
  	}
docs: ------------------------------------------------------------------------