docs: 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>());
      }