code: void setOffsetsToCommit(
  			Map<TopicPartition, OffsetAndMetadata> offsetsToCommit,
  			@Nonnull KafkaCommitCallback commitCallback) {
  
  		// record the work to be committed by the main consumer thread and make sure the consumer notices that
  		if (nextOffsetsToCommit.getAndSet(Tuple2.of(offsetsToCommit, commitCallback)) != null) {
  			log.warn("Committing offsets to Kafka takes longer than the checkpoint interval. " +
  					"Skipping commit of previous offsets because newer complete checkpoint offsets are available. " +
  					"This does not compromise Flink's checkpoint integrity.");
  		}
  
  		// if the consumer is blocked in a poll() or handover operation, wake it up to commit soon
  		handover.wakeupProducer();
  
  		synchronized (consumerReassignmentLock) {
  			if (consumer != null) {
  				consumer.wakeup();
  			} else {
  				// the consumer is currently isolated for partition reassignment;
  				// set this flag so that the wakeup state is restored once the reassignment is complete
  				hasBufferedWakeup = true;
  			}
  		}
  	}
docs: Tells this thread to commit a set of offsets. This method does not block, the committing
  operation will happen asynchronously.
  
  <p>Only one commit operation may be pending at any time. If the committing takes longer than
  the frequency with which this method is called, then some commits may be skipped due to being
  superseded by newer ones.
  
  @param offsetsToCommit The offsets to commit
  @param commitCallback callback when Kafka commit completes