Repository: apache/hadoop
File: hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterClientProtocol.java (Java)

Code before:
  public BatchedEntries<OpenFileEntry> listOpenFiles(long prevId,
      EnumSet<OpenFilesIterator.OpenFilesType> openFilesTypes, String path)
          throws IOException {
    rpcServer.checkOperation(NameNode.OperationCategory.READ, false);
    return null;
  }

  @Override
  public void msync() throws IOException {
    rpcServer.checkOperation(NameNode.OperationCategory.READ, true);
    // Only msync to nameservices with observer reads enabled.
    Set<FederationNamespaceInfo> allNamespaces = namenodeResolver.getNamespaces();
    RemoteMethod method = new RemoteMethod("msync");
    Set<FederationNamespaceInfo> namespacesEligibleForObserverReads = allNamespaces
        .stream()
        .filter(ns -> rpcClient.isNamespaceObserverReadEligible(ns.getNameserviceId()))
        .collect(Collectors.toSet());
    if (namespacesEligibleForObserverReads.isEmpty()) {
      return;
    }
    rpcClient.invokeConcurrent(namespacesEligibleForObserverReads, method);
  }

  @Override
  public void satisfyStoragePolicy(String path) throws IOException {
    storagePolicy.satisfyStoragePolicy(path);
  }

  @Override
  public DatanodeInfo[] getSlowDatanodeReport() throws IOException {
    rpcServer.checkOperation(NameNode.OperationCategory.UNCHECKED);
    return rpcServer.getSlowDatanodeReport(true, 0);
  }

  @Override
  public Path getEnclosingRoot(String src) throws IOException {
    Path mountPath = null;
    if (defaultNameServiceEnabled) {
      mountPath = new Path("/");
    }

    if (subclusterResolver instanceof MountTableResolver) {
      MountTableResolver mountTable = (MountTableResolver) subclusterResolver;
      if (mountTable.getMountPoint(src) != null) {
        mountPath = new Path(mountTable.getMountPoint(src).getSourcePath());
      }
    }

    if (mountPath == null) {
      throw new IOException(String.format("No mount point for %s", src));
    }

Review comment (suggestion):
Hi @kokonguyen191  Here's my question: If a router path `testpath` mounts two services, `ns1` and `ns2`, with `ns1` having 2 files and `ns2` having 5 files, then when `routerclient.listOpenFiles(testpath)` is called, does it first call `RouterClientProtocol.listOpenFiles(0, ALL_OPEN_FILES, testpath)`, returning 4 files (since `minOfMAX=2`, each `ns` returns two files)? Then, because 4 files are returned, does `routerclientProtocol.listOpenFiles(5, ALL_OPEN_FILES, testpath)` only return the 5th file from `ns2`? Would the 3th and 4th files from `ns2` then be missing? I'm unsure if `routerclient.listOpenFiles` is called this way internally.

Code after:

  @Override
  public BatchedEntries<OpenFileEntry> listOpenFiles(long prevId,
      EnumSet<OpenFilesIterator.OpenFilesType> openFilesTypes, String path)
          throws IOException {
    rpcServer.checkOperation(NameNode.OperationCategory.READ, true);
    List<RemoteLocation> locations = rpcServer.getLocationsForPath(path, false, false);
    RemoteMethod method =
        new RemoteMethod("listOpenFiles", new Class<?>[] {long.class, EnumSet.class, String.class},
            prevId, openFilesTypes, new RemoteParam());
    Map<RemoteLocation, BatchedEntries> results =
        rpcClient.invokeConcurrent(locations, method, true, false, -1, BatchedEntries.class);

    // Get the largest inodeIds for each namespace, and the smallest inodeId of them
    // then ignore all entries above this id to keep a consistent prevId for the next listOpenFiles
    long minOfMax = Long.MAX_VALUE;
    for (BatchedEntries nsEntries : results.values()) {
      // Only need to care about namespaces that still have more files to report
      if (!nsEntries.hasMore()) {
        continue;
      }
      long max = 0;
      for (int i = 0; i < nsEntries.size(); i++) {
        max = Math.max(max, ((OpenFileEntry) nsEntries.get(i)).getId());
      }
      minOfMax = Math.min(minOfMax, max);
    }
    // Concatenate all entries into one result, sorted by inodeId
    boolean hasMore = false;
    Map<String, OpenFileEntry> routerEntries = new HashMap<>();
    Map<String, RemoteLocation> resolvedPaths = new HashMap<>();
    for (Map.Entry<RemoteLocation, BatchedEntries> entry : results.entrySet()) {
      BatchedEntries nsEntries = entry.getValue();
      hasMore |= nsEntries.hasMore();
      for (int i = 0; i < nsEntries.size(); i++) {
        OpenFileEntry ofe = (OpenFileEntry) nsEntries.get(i);
        if (ofe.getId() > minOfMax) {
          hasMore = true;
          break;
        }
        RemoteLocation remoteLoc = entry.getKey();
        String routerPath = ofe.getFilePath().replaceFirst(remoteLoc.getDest(), remoteLoc.getSrc());
        OpenFileEntry newEntry =
            new OpenFileEntry(ofe.getId(), routerPath, ofe.getClientName(),
                ofe.getClientMachine());
        // An existing file already resolves to the same path.
        // Resolve according to mount table and keep the best path.
        if (resolvedPaths.containsKey(routerPath)) {
          PathLocation pathLoc = subclusterResolver.getDestinationForPath(routerPath);
          List<String> namespaces = pathLoc.getDestinations().stream().map(
              RemoteLocation::getNameserviceId).collect(