docs: private void emitRecordWithTimestampAndPeriodicWatermark(
  			T record, KafkaTopicPartitionState<KPH> partitionState, long offset, long kafkaEventTimestamp) {
  		@SuppressWarnings("unchecked")
  		final KafkaTopicPartitionStateWithPeriodicWatermarks<T, KPH> withWatermarksState =
  				(KafkaTopicPartitionStateWithPeriodicWatermarks<T, KPH>) partitionState;
  
  		// extract timestamp - this accesses/modifies the per-partition state inside the
  		// watermark generator instance, so we need to lock the access on the
  		// partition state. concurrent access can happen from the periodic emitter
  		final long timestamp;
  		//noinspection SynchronizationOnLocalVariableOrMethodParameter
  		synchronized (withWatermarksState) {
  			timestamp = withWatermarksState.getTimestampForRecord(record, kafkaEventTimestamp);
  		}
  
  		// emit the record with timestamp, using the usual checkpoint lock to guarantee
  		// atomicity of record emission and offset state update
  		synchronized (checkpointLock) {
  			sourceContext.collectWithTimestamp(record, timestamp);
  			partitionState.setOffset(offset);
  		}
  	}