code: public final void commitInternalOffsetsToKafka(
  			Map<KafkaTopicPartition, Long> offsets,
  			@Nonnull KafkaCommitCallback commitCallback) throws Exception {
  		// Ignore sentinels. They might appear here if snapshot has started before actual offsets values
  		// replaced sentinels
  		doCommitInternalOffsetsToKafka(filterOutSentinels(offsets), commitCallback);
  	}
docs: Commits the given partition offsets to the Kafka brokers (or to ZooKeeper for
  older Kafka versions). This method is only ever called when the offset commit mode of
  the consumer is {@link OffsetCommitMode#ON_CHECKPOINTS}.
  
  <p>The given offsets are the internal checkpointed offsets, representing
  the last processed record of each partition. Version-specific implementations of this method
  need to hold the contract that the given offsets must be incremented by 1 before
  committing them, so that committed offsets to Kafka represent "the next record to process".
  
  @param offsets The offsets to commit to Kafka (implementations must increment offsets by 1 before committing).
  @param commitCallback The callback that the user should trigger when a commit request completes or fails.
  @throws Exception This method forwards exceptions.