The way we maintain the lastestPaths in ReplicationSourceManager is broken when sync replication is used

      // Add to latestPaths
      Iterator<Path> iterator = latestPaths.iterator();
      while (iterator.hasNext()) {
        Path path = iterator.next();
        if (path.getName().contains(logPrefix)) {
          iterator.remove();
          break;
        }
      }
      this.latestPaths.add(newLog);