private void seek ( List < TopicPartition > seekList ) { for ( TopicPartition tp : seekList ) { AtomicLong lastCommittedOffset = m_lastCommittedOffSets . get ( ) . get ( tp ) ; if ( lastCommittedOffset != null && lastCommittedOffset . get ( ) > - 1L ) { AtomicLong lastSeeked = m_lastSeekedOffSets . get ( tp ) ; //eliminate duplicate seek if ( lastSeeked != null && lastSeeked . get ( ) == lastCommittedOffset . get ( ) ) { continue ; } m_consumer . seek ( tp , lastCommittedOffset . longValue ( ) ) ; m_lastSeekedOffSets . put ( tp , new AtomicLong ( lastCommittedOffset . get ( ) ) ) ; if ( LOGGER . isDebugEnabled ( ) ) { LOGGER . debug ( "Moves offset for group " + m_config . getGroupId ( ) + " -" + tp + " to " + lastCommittedOffset ) ; } } } }
Move offsets to correct positions for next poll