[SMALLFIX] Fix some bugs in DoraLsCommandIntegrationTest

dora/minicluster/src/main/java/alluxio/master/LocalAlluxioCluster.java

@@ -11,14 +11,20 @@
 
 package alluxio.master;
 
+import alluxio.AlluxioURI;
 import alluxio.ClientContext;
 import alluxio.ConfigurationTestUtils;
 import alluxio.client.block.BlockMasterClient;
+import alluxio.client.block.stream.BlockWorkerClient;
+import alluxio.client.file.DoraCacheFileSystem;
 import alluxio.client.file.FileSystem;
 import alluxio.client.file.FileSystemContext;
 import alluxio.conf.Configuration;
 import alluxio.conf.PropertyKey;
+import alluxio.grpc.GetStatusPRequest;
+import alluxio.grpc.GetStatusPResponse;
 import alluxio.membership.WorkerClusterView;
+import alluxio.resource.CloseableResource;
 import alluxio.util.CommonUtils;
 import alluxio.util.WaitForOptions;
 import alluxio.wire.WorkerInfo;
@@ -36,7 +42,7 @@
 
 /**
  * Local Alluxio cluster for integration tests.
- *
+ * <p>
  * Example to use
  * <pre>
  * // Create a cluster instance
@@ -64,7 +70,7 @@ public LocalAlluxioCluster() {
   }
 
   /**
-   * @param numWorkers the number of workers to run
+   * @param numWorkers   the number of workers to run
    * @param includeProxy weather to include the proxy
    */
   public LocalAlluxioCluster(int numWorkers, boolean includeProxy) {
@@ -184,9 +190,30 @@ protected void waitForMasterServing() throws TimeoutException, InterruptedExcept
   protected void waitForWorkersServing() throws TimeoutException, InterruptedException {
     CommonUtils.waitFor("worker starts serving RPCs", () -> {
       try (FileSystemContext fsContext = FileSystemContext.create()) {
-        WorkerClusterView workers = fsContext.getCachedWorkers();
-        LOG.info("Observed {} workers in the cluster", workers.size());
-        return workers.size() == mNumWorkers;
+        WorkerClusterView workerInfoList = fsContext.getCachedWorkers();
+        if (mNumWorkers != workerInfoList.size()) {
+          return false;
+        }
+        LOG.info("Observed {} workers in the cluster", workerInfoList.size());
+        for (WorkerInfo workerInfo : workerInfoList) {
+          try (CloseableResource<BlockWorkerClient> blockWorkerClient =
+                   fsContext.acquireBlockWorkerClient(workerInfo.getAddress())) {
+            AlluxioURI rootUri = new AlluxioURI("/");
+            FileSystem masterFs = mMaster.getClient();
+            DoraCacheFileSystem doraCacheFs = masterFs.getDoraCacheFileSystem();
+            assert doraCacheFs != null;
+            AlluxioURI uri = doraCacheFs.convertToUfsPath(rootUri);
+            GetStatusPResponse getStatusPResponse = blockWorkerClient.get()
+                .getStatus(GetStatusPRequest.newBuilder().setPath(uri.toString()).build());
+            if (getStatusPResponse == null) {
+              return false;
+            }
+          } catch (IOException ioe) {
+            LOG.error("Failed to connect to worker {}: {}", workerInfo.getAddress(), ioe);
+            return false;
+          }
+        }
+        return true;
       } catch (IOException ioe) {
         LOG.error(ioe.getMessage());
         return false;

dora/tests/integration/src/test/java/alluxio/client/cli/fs/command/DoraLsCommandIntegrationTest.java

@@ -135,7 +135,7 @@ public void testLsR() throws Exception {
     );
   }
 
-  @Ignore("Ignore broken test for now")
+  @Ignore("UfsStatus doesn't support lastAccessTime")
   @Test
   public void testLsWithSortByAccessTime() throws IOException, AlluxioException {
     String oldFile = "/testRoot/oldFile";
@@ -181,7 +181,7 @@ public void testLsWithSortBySizeAndReverse() throws IOException, AlluxioExceptio
     );
   }
 
-  @Ignore("Ignore broken test for now")
+  @Ignore("UfsStatus doesn't support creationTime")
   @Test
   public void testLsWithSortByCreationTime() throws IOException, AlluxioException {
     String oldFile = "/testRoot/oldFile";
@@ -210,6 +210,22 @@ public void testLsWithInvalidSortOption() throws IOException, AlluxioException {
     assertEquals(expected, mOutput.toString());
   }
 
+  @Test
+  public void testLsWithSortByLastModificationTime() throws IOException, AlluxioException {
+    String oldFile = "/testRoot/oldFile";
+    String newFile = "/testRoot/newFile";
+    createByteFileInAlluxio(oldFile, BufferUtils.getIncreasingByteArray(Constants.MB),
+        WritePType.CACHE_THROUGH);
+    createByteFileInAlluxio(newFile, BufferUtils.getIncreasingByteArray(Constants.MB),
+        WritePType.CACHE_THROUGH);
+    assertEquals(0, mFsShell.run("ls", "--sort", "lastModificationTime", "--timestamp",
+        "lastModificationTime", "/testRoot"));
+    checkOutput(
+        ".*FILE " + oldFile,
+        ".*FILE " + newFile
+    );
+  }
+
   private void checkOutput(String... linePatterns) {
     String[] actualLines = mOutput.toString().split("\n");
     assertEquals("Output: ", linePatterns.length, actualLines.length);

dora/tests/integration/src/test/java/alluxio/client/fs/DoraFileSystemIntegrationTest.java

@@ -11,13 +11,13 @@
 
 package alluxio.client.fs;
 
+import static java.lang.Thread.sleep;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThrows;
 
 import alluxio.AlluxioURI;
 import alluxio.Constants;
-import alluxio.annotation.dora.DoraTestTodoItem;
 import alluxio.client.WriteType;
 import alluxio.client.file.FileInStream;
 import alluxio.client.file.FileOutStream;
@@ -42,7 +42,6 @@
 import com.amazonaws.services.s3.AmazonS3ClientBuilder;
 import org.apache.commons.io.IOUtils;
 import org.gaul.s3proxy.junit.S3ProxyRule;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
@@ -52,20 +51,15 @@
 /**
  * Integration tests for Alluxio Client (reuse the {@link LocalAlluxioCluster}).
  */
-@Ignore
-@DoraTestTodoItem(action = DoraTestTodoItem.Action.FIX, owner = "yichuan",
-    comment = "Bring back but not passed, need to fix.")
 public final class DoraFileSystemIntegrationTest extends BaseIntegrationTest {
   @Rule
-  public S3ProxyRule mS3Proxy = S3ProxyRule.builder()
-      .withBlobStoreProvider("transient")
-      .withPort(8001)
-      .withCredentials("_", "_")
-      .build();
+  public S3ProxyRule mS3Proxy =
+      S3ProxyRule.builder().withBlobStoreProvider("transient").withPort(8001)
+          .withCredentials("_", "_").build();
 
   LocalAlluxioClusterResource.Builder mLocalAlluxioClusterResourceBuilder =
-      new LocalAlluxioClusterResource.Builder()
-          .setProperty(PropertyKey.MASTER_PERSISTENCE_CHECKER_INTERVAL_MS, "10ms")
+      new LocalAlluxioClusterResource.Builder().setProperty(
+              PropertyKey.MASTER_PERSISTENCE_CHECKER_INTERVAL_MS, "10ms")
           .setProperty(PropertyKey.MASTER_PERSISTENCE_SCHEDULER_INTERVAL_MS, "10ms")
           .setProperty(PropertyKey.JOB_MASTER_WORKER_HEARTBEAT_INTERVAL, "200ms")
           .setProperty(PropertyKey.USER_BLOCK_SIZE_BYTES_DEFAULT, Constants.MB * 16)
@@ -87,8 +81,7 @@ public final class DoraFileSystemIntegrationTest extends BaseIntegrationTest {
           .setProperty(PropertyKey.S3A_SECRET_KEY, mS3Proxy.getSecretKey())
           .setProperty(PropertyKey.DORA_CLIENT_UFS_FALLBACK_ENABLED, false)
           // current s3 ufs mock don't support setXattr, disable the flag to bypass it
-          .setProperty(PropertyKey.UNDERFS_XATTR_CHANGE_ENABLED, false)
-          .setNumWorkers(2)
+          .setProperty(PropertyKey.UNDERFS_XATTR_CHANGE_ENABLED, false).setNumWorkers(2)
           .setStartCluster(false);
 
   private static final String TEST_BUCKET = "test-bucket";
@@ -107,16 +100,12 @@ private void startCluster(LocalAlluxioClusterResource cluster) throws Exception
     mFileSystem = cluster.get().getClient();
 
     if (mS3Client == null) {
-      mS3Client = AmazonS3ClientBuilder
-          .standard()
-          .withPathStyleAccessEnabled(true)
-          .withCredentials(
+      mS3Client = AmazonS3ClientBuilder.standard().withPathStyleAccessEnabled(true).withCredentials(
               new AWSStaticCredentialsProvider(
                   new BasicAWSCredentials(mS3Proxy.getAccessKey(), mS3Proxy.getSecretKey())))
           .withEndpointConfiguration(
               new AwsClientBuilder.EndpointConfiguration(mS3Proxy.getUri().toString(),
-                  Regions.US_WEST_2.getName()))
-          .build();
+                  Regions.US_WEST_2.getName())).build();
       mS3Client.createBucket(TEST_BUCKET);
     }
   }
@@ -138,28 +127,26 @@ private void writeThenDeleteFromUfs(boolean clientWriteToUFS)
     LocalAlluxioClusterResource clusterResource = mLocalAlluxioClusterResourceBuilder.build();
     startCluster(clusterResource);
 
+//    sleep(1000);
     FileOutStream fos = mFileSystem.createFile(TEST_FILE_URI,
         CreateFilePOptions.newBuilder().setOverwrite(true).build());
     fos.write(TEST_CONTENT.getBytes());
     fos.close();
 
     mS3Client.deleteObject(TEST_BUCKET, TEST_FILE);
-    assertNotNull(mFileSystem.getStatus(TEST_FILE_URI, GetStatusPOptions.newBuilder()
-        .setCommonOptions(optionNoSync())
-        .build()));
+    assertNotNull(mFileSystem.getStatus(TEST_FILE_URI,
+        GetStatusPOptions.newBuilder().setCommonOptions(optionNoSync()).build()));
     try (FileInStream fis = mFileSystem.openFile(TEST_FILE_URI,
         OpenFilePOptions.newBuilder().setCommonOptions(optionNoSync()).build())) {
       String content = IOUtils.toString(fis);
       assertEquals(TEST_CONTENT, content);
     }
 
-    assertThrows(FileDoesNotExistException.class, () ->
-        mFileSystem.getStatus(TEST_FILE_URI, GetStatusPOptions.newBuilder()
-            .setCommonOptions(optionSync()).build()));
+    assertThrows(FileDoesNotExistException.class, () -> mFileSystem.getStatus(TEST_FILE_URI,
+        GetStatusPOptions.newBuilder().setCommonOptions(optionSync()).build()));
 
-    assertThrows(FileDoesNotExistException.class, () ->
-        mFileSystem.getStatus(TEST_FILE_URI, GetStatusPOptions.newBuilder()
-            .setCommonOptions(optionNoSync()).build()));
+    assertThrows(FileDoesNotExistException.class, () -> mFileSystem.getStatus(TEST_FILE_URI,
+        GetStatusPOptions.newBuilder().setCommonOptions(optionNoSync()).build()));
 
     stopCluster(clusterResource);
   }
@@ -176,15 +163,15 @@ private void writeThenUpdateFromUfs(boolean clientWriteToUFS)
     LocalAlluxioClusterResource clusterResource = mLocalAlluxioClusterResourceBuilder.build();
     startCluster(clusterResource);
 
+    sleep(1000);
     FileOutStream fos = mFileSystem.createFile(TEST_FILE_URI,
         CreateFilePOptions.newBuilder().setOverwrite(true).build());
     fos.write(TEST_CONTENT.getBytes());
     fos.close();
 
     mS3Client.putObject(TEST_BUCKET, TEST_FILE, UPDATED_TEST_CONTENT);
-    assertNotNull(mFileSystem.getStatus(TEST_FILE_URI, GetStatusPOptions.newBuilder()
-        .setCommonOptions(optionNoSync())
-        .build()));
+    assertNotNull(mFileSystem.getStatus(TEST_FILE_URI,
+        GetStatusPOptions.newBuilder().setCommonOptions(optionNoSync()).build()));
 
     try (FileInStream fis = mFileSystem.openFile(TEST_FILE_URI,
         OpenFilePOptions.newBuilder().setCommonOptions(optionNoSync()).build())) {
@@ -193,8 +180,8 @@ private void writeThenUpdateFromUfs(boolean clientWriteToUFS)
     }
 
     // This will update/sync metadata from UFS
-    mFileSystem.getStatus(TEST_FILE_URI, GetStatusPOptions.newBuilder()
-        .setCommonOptions(optionSync()).build());
+    mFileSystem.getStatus(TEST_FILE_URI,
+        GetStatusPOptions.newBuilder().setCommonOptions(optionSync()).build());
 
     // metadata is already updated. Even though we are not going to sync metadata in Read(),
     // it should get the latest content.
@@ -208,13 +195,11 @@ private void writeThenUpdateFromUfs(boolean clientWriteToUFS)
   }
 
   private FileSystemMasterCommonPOptions optionNoSync() {
-    return FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(-1)
-        .build();
+    return FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(-1).build();
   }
 
   private FileSystemMasterCommonPOptions optionSync() {
-    return FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(0)
-        .build();
+    return FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(0).build();
   }
 
   /**
@@ -245,6 +230,10 @@ public void testWriteThenDeleteFromUfsNotThrough() throws Exception {
   @Test
   public void testWriteThenUpdateFromUfs() throws Exception {
     writeThenUpdateFromUfs(true);
+  }
+
+  @Test
+  public void testWriteThenUpdateFromUfsNotThrough() throws Exception {
     writeThenUpdateFromUfs(false);
   }
 

dora/tests/integration/src/test/java/alluxio/client/rest/CreateBucketTest.java

@@ -53,11 +53,13 @@ public class CreateBucketTest extends RestApiTest {
           .setProperty(PropertyKey.UNDERFS_S3_ENDPOINT, "localhost:" + UFS_PORT)
           .setProperty(PropertyKey.UNDERFS_S3_ENDPOINT_REGION, "us-west-2")
           .setProperty(PropertyKey.UNDERFS_S3_DISABLE_DNS_BUCKETS, true)
-          .setProperty(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS, "s3://" + TEST_BUCKET)
-          .setProperty(PropertyKey.DORA_CLIENT_UFS_ROOT, "s3://" + TEST_BUCKET)
+          .setProperty(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS, "s3://" + TEST_BUCKET + "/")
+          .setProperty(PropertyKey.DORA_CLIENT_UFS_ROOT, "s3://" + TEST_BUCKET + "/")
           .setProperty(PropertyKey.WORKER_HTTP_SERVER_ENABLED, false)
           .setProperty(PropertyKey.S3A_ACCESS_KEY, mS3Proxy.getAccessKey())
           .setProperty(PropertyKey.S3A_SECRET_KEY, mS3Proxy.getSecretKey())
+          .setProperty(PropertyKey.UNDERFS_XATTR_CHANGE_ENABLED, false)
+          .setProperty(PropertyKey.DORA_CLIENT_UFS_FALLBACK_ENABLED, false)
           .setNumWorkers(1)
           .build();
 

dora/tests/integration/src/test/java/alluxio/client/rest/S3ObjectTest.java

@@ -64,12 +64,13 @@ public class S3ObjectTest extends RestApiTest {
           .setProperty(PropertyKey.UNDERFS_S3_ENDPOINT, "localhost:" + UFS_PORT)
           .setProperty(PropertyKey.UNDERFS_S3_ENDPOINT_REGION, "us-west-2")
           .setProperty(PropertyKey.UNDERFS_S3_DISABLE_DNS_BUCKETS, true)
-          .setProperty(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS, "s3://" + TEST_BUCKET)
-          .setProperty(PropertyKey.DORA_CLIENT_UFS_ROOT, "s3://" + TEST_BUCKET)
+          .setProperty(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS, "s3://" + TEST_BUCKET + "/")
+          .setProperty(PropertyKey.DORA_CLIENT_UFS_ROOT, "s3://" + TEST_BUCKET + "/")
+          .setProperty(PropertyKey.DORA_CLIENT_UFS_FALLBACK_ENABLED, false)
           .setProperty(PropertyKey.WORKER_HTTP_SERVER_ENABLED, false)
           .setProperty(PropertyKey.S3A_ACCESS_KEY, mS3Proxy.getAccessKey())
-          .setProperty(PropertyKey.S3A_SECRET_KEY, mS3Proxy.getSecretKey())
-          .setNumWorkers(1)
+          .setProperty(PropertyKey.UNDERFS_XATTR_CHANGE_ENABLED, false)
+          .setNumWorkers(2)
           .build();
 
   @Before