docs: @Override
      public void onSourceDrained(int partitionId, String tableName) {
          ExportDataSource source;
          synchronized(m_dataSourcesByPartition) {
              Map<String, ExportDataSource> sources = m_dataSourcesByPartition.get(partitionId);
  
              if (sources == null) {
                  if (!m_removingPartitions.contains(partitionId)) {
                       exportLog.error("Could not find export data sources for partition "
                              + partitionId + ". The export cleanup stream is being discarded.");
                  }
                  return;
              }
  
              source = sources.get(tableName);
              if (source == null) {
                  exportLog.warn("Could not find export data source for signature " + partitionId +
                          " name " + tableName + ". The export cleanup stream is being discarded.");
                  return;
              }
  
              // Remove source and partition entry if empty
              sources.remove(tableName);
              if (sources.isEmpty()) {
                  m_dataSourcesByPartition.remove(partitionId);
                  removeMailbox(partitionId);
              }
          }
  
          //Do closing outside the synchronized block. Do not wait on future since
          // we're invoked from the source's executor thread.
          exportLog.info("Drained on unused partition " + partitionId + ": " + source);
          source.closeAndDelete();
      }