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 ) ; } }
Record emission if a timestamp will be attached from an assigner that is also a periodic watermark generator .