private synchronized void getAllPreviousOffsetState ( SourceState state ) { if ( this . doneGettingAllPreviousOffsets ) { return ; } this . previousOffsets . clear ( ) ; this . previousLowWatermarks . clear ( ) ; this . previousExpectedHighWatermarks . clear ( ) ; this . previousOffsetFetchEpochTimes . clear ( ) ; this . previousStartFetchEpochTimes . clear ( ) ; this . previousStopFetchEpochTimes . clear ( ) ; Map < String , Iterable < WorkUnitState > > workUnitStatesByDatasetUrns = state . getPreviousWorkUnitStatesByDatasetUrns ( ) ; if ( ! workUnitStatesByDatasetUrns . isEmpty ( ) && ! ( workUnitStatesByDatasetUrns . size ( ) == 1 && workUnitStatesByDatasetUrns . keySet ( ) . iterator ( ) . next ( ) . equals ( "" ) ) ) { this . isDatasetStateEnabled . set ( true ) ; } for ( WorkUnitState workUnitState : state . getPreviousWorkUnitStates ( ) ) { List < KafkaPartition > partitions = KafkaUtils . getPartitions ( workUnitState ) ; WorkUnit workUnit = workUnitState . getWorkunit ( ) ; MultiLongWatermark watermark = workUnitState . getActualHighWatermark ( MultiLongWatermark . class ) ; MultiLongWatermark previousLowWatermark = workUnit . getLowWatermark ( MultiLongWatermark . class ) ; MultiLongWatermark previousExpectedHighWatermark = workUnit . getExpectedHighWatermark ( MultiLongWatermark . class ) ; Preconditions . checkArgument ( partitions . size ( ) == watermark . size ( ) , String . format ( "Num of partitions doesn't match number of watermarks: partitions=%s, watermarks=%s" , partitions , watermark ) ) ; for ( int i = 0 ; i < partitions . size ( ) ; i ++ ) { KafkaPartition partition = partitions . get ( i ) ; if ( watermark . get ( i ) != ConfigurationKeys . DEFAULT_WATERMARK_VALUE ) { this . previousOffsets . put ( partition , watermark . get ( i ) ) ; } if ( previousLowWatermark . get ( i ) != ConfigurationKeys . DEFAULT_WATERMARK_VALUE ) { this . previousLowWatermarks . put ( partition , previousLowWatermark . get ( i ) ) ; } if ( previousExpectedHighWatermark . get ( i ) != ConfigurationKeys . DEFAULT_WATERMARK_VALUE ) { this . previousExpectedHighWatermarks . put ( partition , previousExpectedHighWatermark . get ( i ) ) ; } this . previousOffsetFetchEpochTimes . put ( partition , KafkaUtils . getPropAsLongFromSingleOrMultiWorkUnitState ( workUnitState , OFFSET_FETCH_EPOCH_TIME , i ) ) ; this . previousStartFetchEpochTimes . put ( partition , KafkaUtils . getPropAsLongFromSingleOrMultiWorkUnitState ( workUnitState , START_FETCH_EPOCH_TIME , i ) ) ; this . previousStopFetchEpochTimes . put ( partition , KafkaUtils . getPropAsLongFromSingleOrMultiWorkUnitState ( workUnitState , STOP_FETCH_EPOCH_TIME , i ) ) ; } } this . doneGettingAllPreviousOffsets = true ; }
this . previousOffsetFetchEpochTimes need to be initialized once