private void calculateTrackers ( Collection < TopicPartition > partitions ) { Map < TopicPartition , CommitTracker > trackers = new HashMap <> ( ) ; trackers . putAll ( m_trackerMap . get ( ) ) ; Map < TopicPartition , AtomicLong > lastCommittedOffSets = new HashMap <> ( ) ; lastCommittedOffSets . putAll ( m_lastCommittedOffSets . get ( ) ) ; boolean newTopicPartition = false ; for ( TopicPartition partition : partitions ) { if ( m_trackerMap . get ( ) . get ( partition ) != null ) { continue ; } newTopicPartition = true ; long startOffset = - 1L ; CommitTracker commitTracker = null ; if ( m_config . getCommitPolicy ( ) == KafkaCommitPolicy . TIME && m_config . getTriggerValue ( ) > 0 ) { commitTracker = new SimpleTracker ( ) ; } else { commitTracker = new DurableTracker ( KafkaConstants . IMPORT_GAP_LEAD , partition . topic ( ) , partition . partition ( ) , m_config . getGroupId ( ) ) ; } trackers . put ( partition , commitTracker ) ; try { OffsetAndMetadata offsetAndMetaData = m_consumer . committed ( partition ) ; startOffset = offsetAndMetaData != null ? offsetAndMetaData . offset ( ) : - 1L ; if ( startOffset > - 1L ) { commitTracker . resetTo ( startOffset ) ; } } catch ( KafkaException e ) { LOGGER . error ( "Failed to read committed offsets for group " + m_config . getGroupId ( ) + partition + " " + e . getMessage ( ) ) ; } lastCommittedOffSets . put ( partition , new AtomicLong ( startOffset ) ) ; m_pauseOffsets . put ( partition , new AtomicLong ( - 1 ) ) ; m_workTrackers . put ( partition , new PendingWorkTracker ( ) ) ; if ( LOGGER . isDebugEnabled ( ) ) { LOGGER . debug ( "Starting offset for group:" + m_config . getGroupId ( ) + ":" + startOffset + " partition:" + partition ) ; } } if ( newTopicPartition ) { m_trackerMap . set ( trackers ) ; m_lastCommittedOffSets . set ( lastCommittedOffSets ) ; } }
add trackers for new topic - partition in this importer