docs: private void removeDataSources(Set<String> doneTables) {
  
          List<ExportDataSource> doneSources = new LinkedList<>();
          synchronized(m_dataSourcesByPartition) {
              for (Iterator<Map<String, ExportDataSource>> it = m_dataSourcesByPartition.values().iterator(); it.hasNext();) {
  
                  Map<String, ExportDataSource> sources = it.next();
                  for (String doneTable : doneTables) {
                      ExportDataSource eds = sources.get(doneTable);
                      if (eds == null) {
                          continue;
                      }
                      doneSources.add(eds);
                      sources.remove(doneTable);
                  }
              }
          }
  
          //Do closings outside the synchronized block and wait for completion.
          List<ListenableFuture<?>> tasks = new ArrayList<ListenableFuture<?>>();
          for (ExportDataSource source : doneSources) {
              exportLog.info("Finished processing " + source);
              tasks.add(source.closeAndDelete());
          }
          try {
              Futures.allAsList(tasks).get();
          } catch (Exception e) {
              //Logging of errors  is done inside the tasks so nothing to do here
              //intentionally not failing if there is an issue with close
              exportLog.error("Error deleting export data sources", e);
          }
      }