Do not use UFSTokenUnavailable in control flow

Do not use UFSTokenUnavailable in control flow

core/common/src/main/java/alluxio/worker/block/BlockLockIdUtil.java

@@ -0,0 +1,49 @@
+/*
+ * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
+ * (the "License"). You may not use this work except in compliance with the License, which is
+ * available at www.apache.org/licenses/LICENSE-2.0
+ *
+ * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied, as more fully set forth in the License.
+ *
+ * See the NOTICE file distributed with this work for information regarding copyright ownership.
+ */
+
+package alluxio.worker.block;
+
+/**
+ * Utils related to block lock ID.
+ */
+public final class BlockLockIdUtil {
+  /** The lock ID that indicates that a UFS block access token is acquired. */
+  public static final long UFS_BLOCK_LOCK_ID = -1;
+  /** The lock ID that indicates that a UFS block access token is not acquired. */
+  public static final long UFS_BLOCK_READ_TOKEN_UNAVAILABLE = -2;
+
+  /** This can be anything that is less than 0 except -1 or -2. */
+  public static final long INVALID_LOCK_ID = -3;
+
+  /**
+   * @param lockId the lock ID
+   * @return true if the lock ID is {@link BlockLockIdUtil#UFS_BLOCK_LOCK_ID}
+   */
+  public static boolean isUfsBlockLockId(long lockId) {
+    return lockId == UFS_BLOCK_LOCK_ID;
+  }
+
+  /**
+   * @param lockId the lock ID
+   * @return true if the lock ID is {@link BlockLockIdUtil#UFS_BLOCK_READ_TOKEN_UNAVAILABLE}
+   */
+  public static boolean isUfsBlockReadTokenUnavailable(long lockId) {
+    return lockId ==  UFS_BLOCK_READ_TOKEN_UNAVAILABLE;
+  }
+
+  /**
+   * @param lockId the lock ID
+   * @return true if the lock ID is a valid Alluxio lock ID
+   */
+  public static boolean isAlluxioBlockLockId(long lockId) {
+    return lockId >= 0;
+  }
+}

core/server/worker/src/main/java/alluxio/worker/block/BlockWorker.java

@@ -379,13 +379,12 @@ void requestSpace(long sessionId, long blockId, long additionalBytes)
    * @param sessionId the session ID
    * @param blockId the block ID
    * @param options the options
+   * @return whether the UFS block is successfully opened
    * @throws BlockAlreadyExistsException if the UFS block already exists in the
    *         {@link UnderFileSystemBlockStore}
-   * @throws UfsBlockAccessTokenUnavailableException if there are too many clients accessing the
-   *         UFS block
    */
-  void openUfsBlock(long sessionId, long blockId, OpenUfsBlockOptions options)
-      throws BlockAlreadyExistsException, UfsBlockAccessTokenUnavailableException;
+  boolean openUfsBlock(long sessionId, long blockId, OpenUfsBlockOptions options)
+      throws BlockAlreadyExistsException;
 
   /**
    * Closes a UFS block for a client session. It also commits the block to Alluxio block store

core/server/worker/src/main/java/alluxio/worker/block/BlockWorkerClientServiceHandler.java

@@ -20,7 +20,6 @@
 import alluxio.WorkerStorageTierAssoc;
 import alluxio.exception.AlluxioException;
 import alluxio.exception.BlockDoesNotExistException;
-import alluxio.exception.UfsBlockAccessTokenUnavailableException;
 import alluxio.exception.UnexpectedAlluxioException;
 import alluxio.exception.WorkerOutOfSpaceException;
 import alluxio.thrift.AlluxioTException;
@@ -168,12 +167,12 @@ public LockBlockResult call() throws AlluxioException {
         }
         // When the block does not exist in Alluxio but exists in UFS, try to open the UFS
         // block.
-        try {
-          mWorker.openUfsBlock(sessionId, blockId, new OpenUfsBlockOptions(options));
+        if (mWorker.openUfsBlock(sessionId, blockId, new OpenUfsBlockOptions(options))) {
           lockId = BlockLockIdUtil.UFS_BLOCK_LOCK_ID;
-        } catch (UfsBlockAccessTokenUnavailableException e) {
+        } else {
           lockId = BlockLockIdUtil.UFS_BLOCK_READ_TOKEN_UNAVAILABLE;
         }
+
         return new LockBlockResult(lockId, "");
       }
 

core/server/worker/src/main/java/alluxio/worker/block/DefaultBlockWorker.java

@@ -441,9 +441,9 @@ public FileInfo getFileInfo(long fileId) throws IOException {
   }
 
   @Override
-  public void openUfsBlock(long sessionId, long blockId, OpenUfsBlockOptions options)
-      throws BlockAlreadyExistsException, UfsBlockAccessTokenUnavailableException {
-    mUnderFileSystemBlockStore.acquireAccess(sessionId, blockId, options);
+  public boolean openUfsBlock(long sessionId, long blockId, OpenUfsBlockOptions options)
+      throws BlockAlreadyExistsException {
+    return mUnderFileSystemBlockStore.acquireAccess(sessionId, blockId, options);
   }
 
   @Override

core/server/worker/src/test/java/alluxio/worker/block/BlockWorkerTest.java

@@ -121,14 +121,9 @@ public void openUnderFileSystemBlock() throws Exception {
 
     long sessionId = 1;
     for (; sessionId < 11; sessionId++) {
-      mBlockWorker.openUfsBlock(sessionId, blockId, openUfsBlockOptions);
-    }
-    try {
-      mBlockWorker.openUfsBlock(sessionId, blockId, openUfsBlockOptions);
-      Assert.fail();
-    } catch (UfsBlockAccessTokenUnavailableException e) {
-      // expected.
+      Assert.assertTrue(mBlockWorker.openUfsBlock(sessionId, blockId, openUfsBlockOptions));
     }
+    Assert.assertFalse(mBlockWorker.openUfsBlock(sessionId, blockId, openUfsBlockOptions));
   }
 
   @Test
@@ -141,10 +136,10 @@ public void closeUnderFileSystemBlock() throws Exception {
 
     long sessionId = 1;
     for (; sessionId < 11; sessionId++) {
-      mBlockWorker.openUfsBlock(sessionId, blockId, openUfsBlockOptions);
+      Assert.assertTrue(mBlockWorker.openUfsBlock(sessionId, blockId, openUfsBlockOptions));
       mBlockWorker.closeUfsBlock(sessionId, blockId);
     }
-    mBlockWorker.openUfsBlock(sessionId, blockId, openUfsBlockOptions);
+    Assert.assertTrue(mBlockWorker.openUfsBlock(sessionId, blockId, openUfsBlockOptions));
   }
 
   /**