docs: protected void emitRecordWithTimestamp(
  			T record, KafkaTopicPartitionState<KPH> partitionState, long offset, long timestamp) throws Exception {
  
  		if (record != null) {
  			if (timestampWatermarkMode == NO_TIMESTAMPS_WATERMARKS) {
  				// fast path logic, in case there are no watermarks generated in the fetcher
  
  				// emit the record, using the checkpoint lock to guarantee
  				// atomicity of record emission and offset state update
  				synchronized (checkpointLock) {
  					sourceContext.collectWithTimestamp(record, timestamp);
  					partitionState.setOffset(offset);
  				}
  			} else if (timestampWatermarkMode == PERIODIC_WATERMARKS) {
  				emitRecordWithTimestampAndPeriodicWatermark(record, partitionState, offset, timestamp);
  			} else {
  				emitRecordWithTimestampAndPunctuatedWatermark(record, partitionState, offset, timestamp);
  			}
  		} else {
  			// if the record is null, simply just update the offset state for partition
  			synchronized (checkpointLock) {
  				partitionState.setOffset(offset);
  			}
  		}
  	}