@Override
  public void listStatus(ListStatusPRequest request,
                         StreamObserver<ListStatusPResponse> responseObserver) {
    LOG.debug("listStatus is called for {}", request.getPath());

    try {
      MultiDimensionalMetricsSystem.META_OPERATION.labelValues("listStatus").inc();
      UfsStatus[] statuses = mWorker.listStatus(request.getPath(), request.getOptions());
      if (statuses == null) {
        responseObserver.onError(
            new NotFoundRuntimeException(String.format("%s Not Found", request.getPath()))
                .toGrpcStatusRuntimeException());
        return;
      }

      ListStatusPResponse.Builder builder = ListStatusPResponse.newBuilder();

      for (int i = 0; i < statuses.length; i++) {
        UfsStatus status = statuses[i];
        String ufsFullPath = PathUtils.concatPath(request.getPath(), status.getName());

        // the list status do not include xattr now. GetAttr will cause some additional overhead.
        // And not every request requires the Xattr. Now only get file xattr in GetStatus.
        alluxio.grpc.FileInfo fi =
            PagedDoraWorker.buildFileInfoFromUfsStatus(mWorker.getCacheUsage(),
                    mWorker.getUfsInstance(ufsFullPath).getUnderFSType(),
                    status, ufsFullPath, null);

        builder.addFileInfos(fi);
        if (builder.getFileInfosCount() == LIST_STATUS_BATCH_SIZE) {
          // Reached the batch size of the reply message. Send it out and create a new one.
          responseObserver.onNext(builder.build());
          builder = ListStatusPResponse.newBuilder();
        }
      }
      if (builder.getFileInfosCount() != 0) {
        // Send out the remaining items if there is any.
        responseObserver.onNext(builder.build());
      }

      responseObserver.onCompleted();
    } catch (Exception e) {
      LOG.error(String.format("Failed to list status of %s: ", request.getPath()), e);
      responseObserver.onError(AlluxioRuntimeException.from(e).toGrpcStatusRuntimeException());
    }
  }
@Test
  public void testListStatusNonExistentDirectory() throws IOException {
    // Setup
    String rootPath = setupTestDirectoryAndFile();

    // Test and Assert
    String nonExistentPath = PathUtils.concatPath(rootPath, "d3");
    TestStreamObserver responseObserver = new TestStreamObserver();
    mRequest = ListStatusPRequest.newBuilder().setOptions(
            alluxio.grpc.ListStatusPOptions.newBuilder().setRecursive(true).build())
        .setPath(nonExistentPath).build();
    assertThrows(RuntimeException.class, () -> mServiceHandler.listStatus(mRequest,
        responseObserver));
  }