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 ) ; } } }
Emits a record attaching a timestamp to it .