code: public FlinkKafkaConsumerBase<T> setStartFromSpecificOffsets(Map<KafkaTopicPartition, Long> specificStartupOffsets) {
  		this.startupMode = StartupMode.SPECIFIC_OFFSETS;
  		this.startupOffsetsTimestamp = null;
  		this.specificStartupOffsets = checkNotNull(specificStartupOffsets);
  		return this;
  	}
docs: Specifies the consumer to start reading partitions from specific offsets, set independently for each partition.
  The specified offset should be the offset of the next record that will be read from partitions.
  This lets the consumer ignore any committed group offsets in Zookeeper / Kafka brokers.
  
  <p>If the provided map of offsets contains entries whose {@link KafkaTopicPartition} is not subscribed by the
  consumer, the entry will be ignored. If the consumer subscribes to a partition that does not exist in the provided
  map of offsets, the consumer will fallback to the default group offset behaviour (see
  {@link FlinkKafkaConsumerBase#setStartFromGroupOffsets()}) for that particular partition.
  
  <p>If the specified offset for a partition is invalid, or the behaviour for that partition is defaulted to group
  offsets but still no group offset could be found for it, then the "auto.offset.reset" behaviour set in the
  configuration properties will be used for the partition
  
  <p>This method does not affect where partitions are read from when the consumer is restored
  from a checkpoint or savepoint. When the consumer is restored from a checkpoint or
  savepoint, only the offsets in the restored state will be used.
  
  @return The consumer object, to allow function chaining.