void initializeGenerationFromCatalog ( CatalogContext catalogContext , final CatalogMap < Connector > connectors , final ExportDataProcessor processor , int hostId , List < Pair < Integer , Integer > > localPartitionsToSites , boolean isCatalogUpdate ) { // Update catalog version so that datasources use this version when propagating acks m_catalogVersion = catalogContext . catalogVersion ; if ( exportLog . isDebugEnabled ( ) ) { exportLog . debug ( "Updating to catalog version : " + m_catalogVersion ) ; } // Collect table names of existing datasources Set < String > currentTables = new HashSet <> ( ) ; synchronized ( m_dataSourcesByPartition ) { for ( Iterator < Map < String , ExportDataSource > > it = m_dataSourcesByPartition . values ( ) . iterator ( ) ; it . hasNext ( ) ; ) { Map < String , ExportDataSource > sources = it . next ( ) ; currentTables . addAll ( sources . keySet ( ) ) ; } } if ( exportLog . isDebugEnabled ( ) ) { exportLog . debug ( "Current tables: " + currentTables ) ; } // Now create datasources based on the catalog (if already present will not be re-created). // Note that we create sources on disabled connectors. Set < Integer > partitionsInUse = localPartitionsToSites . stream ( ) . map ( p -> p . getFirst ( ) ) . collect ( Collectors . toSet ( ) ) ; boolean createdSources = false ; NavigableSet < Table > streams = CatalogUtil . getExportTablesExcludeViewOnly ( connectors ) ; Set < String > exportedTables = new HashSet <> ( ) ; for ( Table stream : streams ) { addDataSources ( stream , hostId , localPartitionsToSites , partitionsInUse , processor , catalogContext . m_genId , isCatalogUpdate ) ; exportedTables . add ( stream . getTypeName ( ) ) ; createdSources = true ; } updateStreamStatus ( exportedTables ) ; // Remove datasources that are not exported anymore for ( String table : exportedTables ) { currentTables . remove ( table ) ; } if ( ! currentTables . isEmpty ( ) ) { removeDataSources ( currentTables ) ; } //Only populate partitions in use if export is actually happening createAckMailboxesIfNeeded ( createdSources ? partitionsInUse : new HashSet < Integer > ( ) ) ; }
Initialize generation from catalog .