Repository: Alluxio/alluxio (language: java)

Issue:
Block location Policy should not create for each file
**Alluxio Version:**
Master branch

**Describe the bug**
Block location Policy should not create for each file. 

**To Reproduce**

Use roundrobin block location policy cannot round robin now, because the policy is created for each file and the index is stored within each policy object.

**Expected behavior**

Each policy should used for all files from a client.

**Urgency**

Urgency.

**Are you planning to fix it**

Yeah

**Additional context**
No



Discussion / hints:

Fix patch:
diff --git a/core/client/fs/src/main/java/alluxio/client/file/BaseFileSystem.java b/core/client/fs/src/main/java/alluxio/client/file/BaseFileSystem.java
index 038e70358624..0a55da818e11 100644
--- a/core/client/fs/src/main/java/alluxio/client/file/BaseFileSystem.java
+++ b/core/client/fs/src/main/java/alluxio/client/file/BaseFileSystem.java
@@ -167,7 +167,7 @@ public FileOutStream createFile(AlluxioURI path, CreateFilePOptions options)
       URIStatus status = client.createFile(path, mergedOptions);
       LOG.debug("Created file {}, options: {}", path.getPath(), mergedOptions);
       OutStreamOptions outStreamOptions =
-          new OutStreamOptions(mergedOptions, mFsContext.getClientContext(),
+          new OutStreamOptions(mergedOptions, mFsContext,
               mFsContext.getPathConf(path));
       outStreamOptions.setUfsPath(status.getUfsPath());
       outStreamOptions.setMountId(status.getMountId());
@@ -407,7 +407,7 @@ public FileInStream openFile(URIStatus status, OpenFilePOptions options)
     AlluxioConfiguration conf = mFsContext.getPathConf(path);
     OpenFilePOptions mergedOptions = FileSystemOptions.openFileDefaults(conf)
         .toBuilder().mergeFrom(options).build();
-    InStreamOptions inStreamOptions = new InStreamOptions(status, mergedOptions, conf);
+    InStreamOptions inStreamOptions = new InStreamOptions(status, mergedOptions, conf, mFsContext);
     return new AlluxioFileInStream(status, inStreamOptions, mFsContext);
   }
 
diff --git a/core/client/fs/src/main/java/alluxio/client/file/FileSystemContext.java b/core/client/fs/src/main/java/alluxio/client/file/FileSystemContext.java
index 0927dfa216cf..7025ffa61bdf 100644
--- a/core/client/fs/src/main/java/alluxio/client/file/FileSystemContext.java
+++ b/core/client/fs/src/main/java/alluxio/client/file/FileSystemContext.java
@@ -18,6 +18,7 @@
 import alluxio.client.block.BlockMasterClient;
 import alluxio.client.block.BlockMasterClientPool;
 import alluxio.client.block.BlockWorkerInfo;
+import alluxio.client.block.policy.BlockLocationPolicy;
 import alluxio.client.block.stream.BlockWorkerClient;
 import alluxio.client.block.stream.BlockWorkerClientPool;
 import alluxio.client.file.FileSystemContextReinitializer.ReinitBlockerResource;
@@ -58,6 +59,7 @@
 import java.net.SocketAddress;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -164,6 +166,8 @@ public class FileSystemContext implements Closeable {
   @GuardedBy("this")
   private final RefreshPolicy mWorkerRefreshPolicy;
 
+  private final Map<Class, BlockLocationPolicy> mBlockLocationPolicyMap;
+
   /**
    * Creates a {@link FileSystemContext} with an empty subject, default config
    * and a null local block worker.
@@ -251,6 +255,7 @@ private FileSystemContext(AlluxioConfiguration conf, @Nullable BlockWorker block
         new TimeoutRefresh(conf.getMs(PropertyKey.USER_WORKER_LIST_REFRESH_INTERVAL));
     LOG.debug("Created context with id: {}, with local block worker: {}",
         mId, mBlockWorker != null);
+    mBlockLocationPolicyMap = new ConcurrentHashMap();
   }
 
   /**
@@ -696,6 +701,32 @@ private List<WorkerNetAddress> getWorkerAddresses() throws IOException {
     return localWorkerNetAddresses.isEmpty() ? workerNetAddresses : localWorkerNetAddresses;
   }
 
+  /**
+   * Gets the readBlockLocationPolicy.
+   *
+   * @param alluxioConf Alluxio configuration
+   *
+   * @return the readBlockLocationPolicy
+   */
+  public BlockLocationPolicy getReadBlockLocationPolicy(AlluxioConfiguration alluxioConf) {
+    return mBlockLocationPolicyMap.computeIfAbsent(
+        alluxioConf.getClass(PropertyKey.USER_UFS_BLOCK_READ_LOCATION_POLICY),
+        pc -> BlockLocationPolicy.Factory.create(pc, alluxioConf));
+  }
+
+  /**
+   * Gets the writeBlockLocationPolicy.
+   *
+   * @param alluxioConf Alluxio configuration
+   *
+   * @return the writeBlockLocationPolicy
+   */
+  public BlockLocationPolicy getWriteBlockLocationPolicy(AlluxioConfiguration alluxioConf) {
+    return mBlockLocationPolicyMap.computeIfAbsent(
+        alluxioConf.getClass(PropertyKey.USER_BLOCK_WRITE_LOCATION_POLICY),
+        pc -> BlockLocationPolicy.Factory.create(pc, alluxioConf));
+  }
+
   /**
    * Key for block worker client pools. This requires both the worker address and the username, so
    * that block workers are created for different users.
diff --git a/core/client/fs/src/main/java/alluxio/client/file/options/InStreamOptions.java b/core/client/fs/src/main/java/alluxio/client/file/options/InStreamOptions.java
index 239c33d29ae8..fbf22ee99797 100644
--- a/core/client/fs/src/main/java/alluxio/client/file/options/InStreamOptions.java
+++ b/core/client/fs/src/main/java/alluxio/client/file/options/InStreamOptions.java
@@ -13,6 +13,7 @@
 
 import alluxio.client.ReadType;
 import alluxio.client.block.policy.BlockLocationPolicy;
+import alluxio.client.file.FileSystemContext;
 import alluxio.client.file.URIStatus;
 import alluxio.conf.AlluxioConfiguration;
 import alluxio.conf.PropertyKey;
@@ -29,6 +30,7 @@
 import com.google.common.base.Objects;
 import com.google.common.base.Preconditions;
 
+import javax.annotation.Nonnull;
 import javax.annotation.concurrent.NotThreadSafe;
 
 /**
@@ -51,8 +53,21 @@ public final class InStreamOptions {
    * @param status the file to create the options for
    * @param alluxioConf Alluxio configuration
    */
-  public InStreamOptions(URIStatus status, AlluxioConfiguration alluxioConf) {
-    this(status, FileSystemOptions.openFileDefaults(alluxioConf), alluxioConf);
+  public InStreamOptions(URIStatus status, @Nonnull AlluxioConfiguration alluxioConf) {
+    this(status, FileSystemOptions.openFileDefaults(alluxioConf), alluxioConf,
+        FileSystemContext.create(alluxioConf));
+  }
+
+  /**
+   * Creates with the default {@link OpenFilePOptions}.
+   *
+   * @param status the file to create the options for
+   * @param alluxioConf Alluxio configuration
+   * @param context the file system context
+   */
+  public InStreamOptions(URIStatus status, @Nonnull AlluxioConfiguration alluxioConf,
+      @Nonnull FileSystemContext context) {
+    this(status, FileSystemOptions.openFileDefaults(alluxioConf), alluxioConf, context);
   }
 
   /**
@@ -60,9 +75,11 @@ public InStreamOptions(URIStatus status, AlluxioConfiguration alluxioConf) {
    * @param status URI status
    * @param options OpenFile options
    * @param alluxioConf Alluxio configuration
+   * @param context the file system context
    */
-  public InStreamOptions(URIStatus status, OpenFilePOptions options,
-      AlluxioConfiguration alluxioConf) {
+  public InStreamOptions(URIStatus status, @Nonnull OpenFilePOptions options,
+      @Nonnull AlluxioConfiguration alluxioConf, @Nonnull FileSystemContext context) {
+    Preconditions.checkNotNull(context);
     // Create OpenOptions builder with default options.
     OpenFilePOptions.Builder openOptionsBuilder = OpenFilePOptions.newBuilder()
         .setReadType(alluxioConf.getEnum(PropertyKey.USER_FILE_READ_TYPE_DEFAULT, ReadType.class)
@@ -74,8 +91,7 @@ public InStreamOptions(URIStatus status, OpenFilePOptions options,
 
     mStatus = status;
     mProtoOptions = openOptions;
-    mUfsReadLocationPolicy = BlockLocationPolicy.Factory.create(
-        alluxioConf.getClass(PropertyKey.USER_UFS_BLOCK_READ_LOCATION_POLICY), alluxioConf);
+    mUfsReadLocationPolicy = context.getReadBlockLocationPolicy(alluxioConf);
     mPositionShort = false;
   }
 
@@ -93,7 +109,6 @@ public OpenFilePOptions getOptions() {
    */
   @VisibleForTesting
   public void setUfsReadLocationPolicy(BlockLocationPolicy ufsReadLocationPolicy) {
-
     mUfsReadLocationPolicy = Preconditions.checkNotNull(ufsReadLocationPolicy);
   }
 
diff --git a/core/client/fs/src/main/java/alluxio/client/file/options/OutStreamOptions.java b/core/client/fs/src/main/java/alluxio/client/file/options/OutStreamOptions.java
index 9303edebb191..36a574f4f044 100644
--- a/core/client/fs/src/main/java/alluxio/client/file/options/OutStreamOptions.java
+++ b/core/client/fs/src/main/java/alluxio/client/file/options/OutStreamOptions.java
@@ -11,12 +11,12 @@
 
 package alluxio.client.file.options;
 
-import alluxio.ClientContext;
 import alluxio.annotation.PublicApi;
 import alluxio.client.AlluxioStorageType;
 import alluxio.client.UnderStorageType;
 import alluxio.client.WriteType;
 import alluxio.client.block.policy.BlockLocationPolicy;
+import alluxio.client.file.FileSystemContext;
 import alluxio.conf.AlluxioConfiguration;
 import alluxio.conf.PropertyKey;
 import alluxio.grpc.CreateFilePOptions;
@@ -62,7 +62,8 @@ public final class OutStreamOptions {
    * @param alluxioConf the Alluxio configuration
    * @return the default {@link OutStreamOptions}
    */
-  public static OutStreamOptions defaults(ClientContext context, AlluxioConfiguration alluxioConf) {
+  public static OutStreamOptions defaults(FileSystemContext context,
+      AlluxioConfiguration alluxioConf) {
     return new OutStreamOptions(context, alluxioConf);
   }
 
@@ -70,7 +71,7 @@ public static OutStreamOptions defaults(ClientContext context, AlluxioConfigurat
    * @param context Alluxio client context
    * @return the default {@link OutStreamOptions}
    */
-  public static OutStreamOptions defaults(ClientContext context) {
+  public static OutStreamOptions defaults(FileSystemContext context) {
     return new OutStreamOptions(context, context.getClusterConf());
   }
 
@@ -82,7 +83,7 @@ public static OutStreamOptions defaults(ClientContext context) {
    * @param alluxioConf the Alluxio configuration
    * @throws Exception if {@link BlockLocationPolicy} can't be loaded
    */
-  public OutStreamOptions(CreateFilePOptions options, ClientContext context,
+  public OutStreamOptions(CreateFilePOptions options, FileSystemContext context,
       AlluxioConfiguration alluxioConf) {
     this(context, alluxioConf);
     if (options.hasCommonOptions()) {
@@ -112,24 +113,16 @@ public OutStreamOptions(CreateFilePOptions options, ClientContext context,
     if (options.hasWriteType()) {
       mWriteType = WriteType.fromProto(options.getWriteType());
     }
-    try {
-      mLocationPolicy = BlockLocationPolicy.Factory.create(
-          alluxioConf.getClass(PropertyKey.USER_BLOCK_WRITE_LOCATION_POLICY), alluxioConf);
-    } catch (Exception e) {
-      throw new RuntimeException(e);
-    }
   }
 
-  private OutStreamOptions(ClientContext context, AlluxioConfiguration alluxioConf) {
+  private OutStreamOptions(FileSystemContext context, AlluxioConfiguration alluxioConf) {
     mCommonOptions = FileSystemOptions.commonDefaults(alluxioConf);
     mBlockSizeBytes = alluxioConf.getBytes(PropertyKey.USER_BLOCK_SIZE_BYTES_DEFAULT);
-    mLocationPolicy = BlockLocationPolicy.Factory.create(
-        alluxioConf.getClass(PropertyKey.USER_BLOCK_WRITE_LOCATION_POLICY),
-        alluxioConf);
+    mLocationPolicy = context.getWriteBlockLocationPolicy(alluxioConf);
     mWriteTier = alluxioConf.getInt(PropertyKey.USER_FILE_WRITE_TIER_DEFAULT);
     mWriteType = alluxioConf.getEnum(PropertyKey.USER_FILE_WRITE_TYPE_DEFAULT, WriteType.class);
     try {
-      mOwner = context.getUserState().getUser().getName();
+      mOwner = context.getClientContext().getUserState().getUser().getName();
       mGroup = CommonUtils.getPrimaryGroupName(mOwner, context.getClusterConf());
     } catch (IOException e) {
       mOwner = "";
diff --git a/job/server/src/main/java/alluxio/job/util/JobUtils.java b/job/server/src/main/java/alluxio/job/util/JobUtils.java
index 17765db00d50..c3fba4c2798b 100644
--- a/job/server/src/main/java/alluxio/job/util/JobUtils.java
+++ b/job/server/src/main/java/alluxio/job/util/JobUtils.java
@@ -163,12 +163,12 @@ public static void loadBlock(URIStatus status, FileSystemContext context, long b
     OpenFilePOptions openOptions =
         OpenFilePOptions.newBuilder().setReadType(ReadPType.NO_CACHE).build();
 
-    InStreamOptions inOptions = new InStreamOptions(status, openOptions, conf);
+    InStreamOptions inOptions = new InStreamOptions(status, openOptions, conf, context);
     // Set read location policy always to local first for loading blocks for job tasks
     inOptions.setUfsReadLocationPolicy(BlockLocationPolicy.Factory.create(
         LocalFirstPolicy.class, conf));
 
-    OutStreamOptions outOptions = OutStreamOptions.defaults(context.getClientContext());
+    OutStreamOptions outOptions = OutStreamOptions.defaults(context);
     outOptions.setMediumType(medium);
     // Set write location policy always to local first for loading blocks for job tasks
     outOptions.setLocationPolicy(BlockLocationPolicy.Factory.create(
@@ -199,7 +199,7 @@ private static void loadThroughCacheRequest(URIStatus status, FileSystemContext
     BlockStoreClient blockStore = BlockStoreClient.create(context);
     OpenFilePOptions openOptions =
         OpenFilePOptions.newBuilder().setReadType(ReadPType.CACHE).build();
-    InStreamOptions inOptions = new InStreamOptions(status, openOptions, conf);
+    InStreamOptions inOptions = new InStreamOptions(status, openOptions, conf, context);
     BlockLocationPolicy policy =
         BlockLocationPolicy.Factory.create(LocalFirstPolicy.class, conf);
     inOptions.setUfsReadLocationPolicy(policy);
@@ -231,7 +231,7 @@ private static void loadThroughRead(URIStatus status, FileSystemContext context,
     BlockStoreClient blockStore = BlockStoreClient.create(context);
     OpenFilePOptions openOptions =
         OpenFilePOptions.newBuilder().setReadType(ReadPType.CACHE).build();
-    InStreamOptions inOptions = new InStreamOptions(status, openOptions, conf);
+    InStreamOptions inOptions = new InStreamOptions(status, openOptions, conf, context);
     inOptions.setUfsReadLocationPolicy(BlockLocationPolicy.Factory.create(
         LocalFirstPolicy.class, conf));
     BlockInfo info = Preconditions.checkNotNull(status.getBlockInfo(blockId));
diff --git a/shell/src/main/java/alluxio/cli/fs/command/LoadCommand.java b/shell/src/main/java/alluxio/cli/fs/command/LoadCommand.java
index beb256e225cc..68a22dd2effc 100644
--- a/shell/src/main/java/alluxio/cli/fs/command/LoadCommand.java
+++ b/shell/src/main/java/alluxio/cli/fs/command/LoadCommand.java
@@ -143,7 +143,7 @@ private void runLoadTask(AlluxioURI filePath, URIStatus status, boolean local)
         dataSource = dataSourceAndType.getFirst();
       }
       Protocol.OpenUfsBlockOptions openUfsBlockOptions =
-          new InStreamOptions(status, options, conf).getOpenUfsBlockOptions(blockId);
+          new InStreamOptions(status, options, conf, mFsContext).getOpenUfsBlockOptions(blockId);
       cacheBlock(blockId, dataSource, status, openUfsBlockOptions);
     }
   }


Test patch:
diff --git a/core/client/fs/src/test/java/alluxio/client/block/BlockStoreClientTest.java b/core/client/fs/src/test/java/alluxio/client/block/BlockStoreClientTest.java
index 2572c314d462..a940ed8ecb7b 100644
--- a/core/client/fs/src/test/java/alluxio/client/block/BlockStoreClientTest.java
+++ b/core/client/fs/src/test/java/alluxio/client/block/BlockStoreClientTest.java
@@ -167,6 +167,13 @@ public void before() throws Exception {
         .thenReturn(new DummyCloseableResource<>(mMasterClient));
     when(mContext.getClientContext()).thenReturn(mClientContext);
     when(mContext.getClusterConf()).thenReturn(S_CONF);
+    when(mContext.getReadBlockLocationPolicy(any(AlluxioConfiguration.class)))
+        .thenAnswer((Answer) invocation -> {
+          AlluxioConfiguration conf =
+              invocation.getArgument(0, AlluxioConfiguration.class);
+          return BlockLocationPolicy.Factory.create(
+              conf.getClass(PropertyKey.USER_UFS_BLOCK_READ_LOCATION_POLICY), conf);
+        });
 
     mBlockStore = new BlockStoreClient(mContext,
         TieredIdentityFactory.fromString("node=" + WORKER_HOSTNAME_LOCAL, S_CONF));
@@ -186,7 +193,7 @@ public void before() throws Exception {
   @Test
   public void getOutStreamUsingLocationPolicy() {
     OutStreamOptions options =
-        OutStreamOptions.defaults(mClientContext).setWriteType(WriteType.MUST_CACHE)
+        OutStreamOptions.defaults(mContext).setWriteType(WriteType.MUST_CACHE)
             .setLocationPolicy((workerOptions) -> {
               throw new RuntimeException("policy threw exception");
             });
@@ -196,7 +203,7 @@ public void getOutStreamUsingLocationPolicy() {
   @Test
   public void getOutStreamMissingLocationPolicy() {
     OutStreamOptions options =
-        OutStreamOptions.defaults(mClientContext).setBlockSizeBytes(BLOCK_LENGTH)
+        OutStreamOptions.defaults(mContext).setBlockSizeBytes(BLOCK_LENGTH)
             .setWriteType(WriteType.MUST_CACHE).setLocationPolicy(null);
     Exception e = assertThrows(NullPointerException.class, () ->
         mBlockStore.getOutStream(BLOCK_ID, BLOCK_LENGTH, options));
@@ -208,7 +215,7 @@ public void getOutStreamMissingLocationPolicy() {
   public void getOutStreamNoWorker() {
     OutStreamOptions options =
         OutStreamOptions
-            .defaults(mClientContext)
+            .defaults(mContext)
             .setBlockSizeBytes(BLOCK_LENGTH)
             .setWriteType(WriteType.MUST_CACHE)
             .setLocationPolicy(
@@ -232,7 +239,7 @@ public void getOutStreamLocal() throws Exception {
           return mStreamObserver;
         });
 
-    OutStreamOptions options = OutStreamOptions.defaults(mClientContext)
+    OutStreamOptions options = OutStreamOptions.defaults(mContext)
         .setBlockSizeBytes(BLOCK_LENGTH).setLocationPolicy(
             new MockBlockLocationPolicyTest(Lists.newArrayList(WORKER_NET_ADDRESS_LOCAL)))
             .setWriteType(WriteType.MUST_CACHE);
@@ -245,7 +252,7 @@ public void getOutStreamRemote() throws Exception {
     WorkerNetAddress worker1 = new WorkerNetAddress().setHost("worker1");
     WorkerNetAddress worker2 = new WorkerNetAddress().setHost("worker2");
     OutStreamOptions options =
-        OutStreamOptions.defaults(mClientContext).setBlockSizeBytes(BLOCK_LENGTH)
+        OutStreamOptions.defaults(mContext).setBlockSizeBytes(BLOCK_LENGTH)
             .setLocationPolicy(new MockBlockLocationPolicyTest(Arrays.asList(worker1, worker2)))
             .setWriteType(WriteType.MUST_CACHE);
     BlockOutStream stream1 = mBlockStore.getOutStream(BLOCK_ID, BLOCK_LENGTH, options);
@@ -271,7 +278,7 @@ public void getOutStreamWithReplicated() throws Exception {
         .newArrayList(new BlockWorkerInfo(WORKER_NET_ADDRESS_LOCAL, -1, -1),
             new BlockWorkerInfo(WORKER_NET_ADDRESS_REMOTE, -1, -1)));
     OutStreamOptions options =
-        OutStreamOptions.defaults(mClientContext).setBlockSizeBytes(BLOCK_LENGTH).setLocationPolicy(
+        OutStreamOptions.defaults(mContext).setBlockSizeBytes(BLOCK_LENGTH).setLocationPolicy(
             new MockBlockLocationPolicyTest(
                 Lists.newArrayList(WORKER_NET_ADDRESS_LOCAL, WORKER_NET_ADDRESS_REMOTE)))
             .setWriteType(WriteType.MUST_CACHE).setReplicationMin(2);
@@ -291,7 +298,7 @@ public void getInStreamUfsMockLocaltion() throws Exception {
           .setBlockIds(Collections.singletonList(0L))
           .setFileBlockInfos(Collections.singletonList(new FileBlockInfo().setBlockInfo(info))));
       OpenFilePOptions readOptions = OpenFilePOptions.newBuilder().build();
-      InStreamOptions options = new InStreamOptions(dummyStatus, readOptions, S_CONF);
+      InStreamOptions options = new InStreamOptions(dummyStatus, readOptions, S_CONF, mContext);
       ((MockBlockLocationPolicyTest) options.getUfsReadLocationPolicy())
           .setHosts(Arrays.asList(worker1, worker2));
       when(mMasterClient.getBlockInfo(BLOCK_ID)).thenReturn(new BlockInfo());
@@ -315,7 +322,7 @@ public void getInStreamUfsLocalFirst() throws Exception {
         new URIStatus(new FileInfo().setPersisted(true).setBlockIds(Collections.singletonList(0L))
             .setFileBlockInfos(Collections.singletonList(new FileBlockInfo().setBlockInfo(info))));
     OpenFilePOptions readOptions = OpenFilePOptions.newBuilder().build();
-    InStreamOptions options = new InStreamOptions(dummyStatus, readOptions, S_CONF);
+    InStreamOptions options = new InStreamOptions(dummyStatus, readOptions, S_CONF, mContext);
     when(mMasterClient.getBlockInfo(BLOCK_ID)).thenReturn(new BlockInfo());
     when(mContext.getCachedWorkers()).thenReturn(
         Lists.newArrayList(new BlockWorkerInfo(remote, 100, 0),
@@ -333,7 +340,7 @@ public void getInStreamNoWorkers() throws Exception {
         new URIStatus(new FileInfo().setPersisted(true).setBlockIds(Collections.singletonList(0L)));
     InStreamOptions options =
         new InStreamOptions(dummyStatus, FileSystemOptions.openFileDefaults(S_CONF),
-            S_CONF);
+            S_CONF, mContext);
     when(mMasterClient.getBlockInfo(BLOCK_ID)).thenReturn(new BlockInfo());
     when(mContext.getCachedWorkers()).thenReturn(Collections.emptyList());
     Exception e = assertThrows(UnavailableException.class, () ->
@@ -346,7 +353,8 @@ public void getInStreamMissingBlock() throws Exception {
     URIStatus dummyStatus = new URIStatus(
         new FileInfo().setPersisted(false).setBlockIds(Collections.singletonList(0L)));
     InStreamOptions options =
-        new InStreamOptions(dummyStatus, FileSystemOptions.openFileDefaults(S_CONF), S_CONF);
+        new InStreamOptions(dummyStatus, FileSystemOptions.openFileDefaults(S_CONF),
+            S_CONF, mContext);
     when(mMasterClient.getBlockInfo(BLOCK_ID)).thenReturn(new BlockInfo());
     Exception e = assertThrows(UnavailableException.class, () ->
         mBlockStore.getInStream(BLOCK_ID, options).getAddress());
@@ -376,8 +384,9 @@ public void getInStreamLocal() throws Exception {
 
     when(mMasterClient.getBlockInfo(BLOCK_ID)).thenReturn(info);
     assertEquals(local, mBlockStore.getInStream(BLOCK_ID, new InStreamOptions(
-        new URIStatus(new FileInfo().setBlockIds(Lists.newArrayList(BLOCK_ID))), S_CONF))
-                            .getAddress());
+        new URIStatus(new FileInfo().setBlockIds(Lists.newArrayList(BLOCK_ID))),
+            S_CONF, mContext))
+        .getAddress());
   }
 
   @Test
@@ -394,7 +403,8 @@ public void getInStreamRemote() throws Exception {
     Set<WorkerNetAddress> results = new HashSet<>();
     for (int i = 0; i < 40; i++) {
       results.add(mBlockStore.getInStream(BLOCK_ID, new InStreamOptions(
-          new URIStatus(new FileInfo().setBlockIds(Lists.newArrayList(BLOCK_ID))), S_CONF))
+          new URIStatus(new FileInfo().setBlockIds(Lists.newArrayList(BLOCK_ID))),
+              S_CONF, mContext))
           .getAddress());
     }
     assertEquals(Sets.newHashSet(remote1, remote2), results);
@@ -414,7 +424,8 @@ public void getInStreamProcessLocal() throws Exception {
     when(mContext.getProcessLocalWorker()).thenReturn(blockWorker);
 
     BlockInStream stream = mBlockStore.getInStream(BLOCK_ID, new InStreamOptions(
-        new URIStatus(new FileInfo().setBlockIds(Lists.newArrayList(BLOCK_ID))), S_CONF));
+        new URIStatus(new FileInfo().setBlockIds(Lists.newArrayList(BLOCK_ID))),
+        S_CONF, mContext));
     assertEquals(local, stream.getAddress());
     assertEquals(BlockWorkerDataReader.Factory.class.getName(),
         stream.getDataReaderFactory().getClass().getName());
@@ -429,7 +440,7 @@ public void getInStreamUfsProcessLocal() throws Exception {
         new URIStatus(new FileInfo().setPersisted(true).setBlockIds(Collections.singletonList(0L))
             .setFileBlockInfos(Collections.singletonList(new FileBlockInfo().setBlockInfo(info))));
     OpenFilePOptions readOptions = OpenFilePOptions.newBuilder().build();
-    InStreamOptions options = new InStreamOptions(dummyStatus, readOptions, S_CONF);
+    InStreamOptions options = new InStreamOptions(dummyStatus, readOptions, S_CONF, mContext);
     when(mMasterClient.getBlockInfo(BLOCK_ID)).thenReturn(new BlockInfo());
     when(mContext.getCachedWorkers()).thenReturn(
         Lists.newArrayList(new BlockWorkerInfo(remote, 100, 0),
@@ -529,7 +540,8 @@ public void getInStreamInAlluxioWhenCreateStreamIsFailed() throws Exception {
         .getArgument(0, GetWorkerOptions.class).getBlockWorkerInfos().iterator().next()
         .getNetAddress());
     InStreamOptions options =
-        new InStreamOptions(dummyStatus, FileSystemOptions.openFileDefaults(S_CONF), S_CONF);
+        new InStreamOptions(dummyStatus, FileSystemOptions.openFileDefaults(S_CONF),
+            S_CONF, mContext);
     options.setUfsReadLocationPolicy(mockPolicy);
     when(mMasterClient.getBlockInfo(BLOCK_ID)).thenReturn(info);
     when(mContext.getCachedWorkers()).thenReturn(
@@ -569,7 +581,8 @@ private void testGetInStreamFallback(int workerCount, boolean isPersisted, int[]
         .getArgument(0, GetWorkerOptions.class).getBlockWorkerInfos().iterator().next()
         .getNetAddress()));
     InStreamOptions options =
-        new InStreamOptions(dummyStatus, FileSystemOptions.openFileDefaults(S_CONF), S_CONF);
+        new InStreamOptions(dummyStatus, FileSystemOptions.openFileDefaults(S_CONF),
+            S_CONF, mContext);
     options.setUfsReadLocationPolicy(mockPolicy);
     when(mMasterClient.getBlockInfo(BLOCK_ID)).thenReturn(info);
     when(mContext.getCachedWorkers()).thenReturn(Arrays.stream(workers)
diff --git a/core/client/fs/src/test/java/alluxio/client/block/stream/BlockInStreamTest.java b/core/client/fs/src/test/java/alluxio/client/block/stream/BlockInStreamTest.java
index 5bb13497f22d..787b216cca2e 100644
--- a/core/client/fs/src/test/java/alluxio/client/block/stream/BlockInStreamTest.java
+++ b/core/client/fs/src/test/java/alluxio/client/block/stream/BlockInStreamTest.java
@@ -86,7 +86,7 @@ public void before() throws Exception {
     when(mMockContext.getClusterConf()).thenReturn(mConf);
     mInfo = new BlockInfo().setBlockId(1);
     mOptions = new InStreamOptions(new URIStatus(new FileInfo().setBlockIds(Collections
-        .singletonList(1L))), mConf);
+        .singletonList(1L))), mConf, mMockContext);
   }
 
   @Test
diff --git a/core/client/fs/src/test/java/alluxio/client/block/stream/GrpcDataWriterTest.java b/core/client/fs/src/test/java/alluxio/client/block/stream/GrpcDataWriterTest.java
index c922eacec312..5fb326da9dcf 100644
--- a/core/client/fs/src/test/java/alluxio/client/block/stream/GrpcDataWriterTest.java
+++ b/core/client/fs/src/test/java/alluxio/client/block/stream/GrpcDataWriterTest.java
@@ -196,7 +196,7 @@ public void closedBlockWorkerClientTest() throws IOException {
     mConf.set(PropertyKey.USER_STREAMING_WRITER_CLOSE_TIMEOUT, "1");
     GrpcDataWriter writer = GrpcDataWriter.create(context, mAddress, BLOCK_ID, 0,
         RequestType.ALLUXIO_BLOCK,
-        OutStreamOptions.defaults(mClientContext).setWriteTier(0));
+        OutStreamOptions.defaults(mContext).setWriteTier(0));
     verify(resource, times(0)).close();
     verifyWriteRequests(mClient, 0, 0);
     writer.close();
@@ -213,7 +213,7 @@ private DataWriter create(long length) throws Exception {
     DataWriter writer =
         GrpcDataWriter.create(mContext, mAddress, BLOCK_ID, length,
             RequestType.ALLUXIO_BLOCK,
-            OutStreamOptions.defaults(mClientContext).setWriteTier(TIER));
+            OutStreamOptions.defaults(mContext).setWriteTier(TIER));
     return writer;
   }
 
diff --git a/core/client/fs/src/test/java/alluxio/client/block/stream/LocalFileDataWriterTest.java b/core/client/fs/src/test/java/alluxio/client/block/stream/LocalFileDataWriterTest.java
index 1e6994c29269..0848036bc789 100644
--- a/core/client/fs/src/test/java/alluxio/client/block/stream/LocalFileDataWriterTest.java
+++ b/core/client/fs/src/test/java/alluxio/client/block/stream/LocalFileDataWriterTest.java
@@ -91,7 +91,7 @@ public void after() throws Exception {
   @Test
   public void streamCancelled() throws Exception {
     LocalFileDataWriter writer = LocalFileDataWriter.create(mContext, mAddress, BLOCK_ID,
-        128 /* unused */, OutStreamOptions.defaults(mClientContext));
+        128 /* unused */, OutStreamOptions.defaults(mContext));
 
     // Cancel stream before cancelling the writer
     PowerMockito.when(mStream.isCanceled()).thenReturn(true);
@@ -108,7 +108,7 @@ public void streamCancelled() throws Exception {
   @Test
   public void streamClosed() throws Exception {
     LocalFileDataWriter writer = LocalFileDataWriter.create(mContext, mAddress, BLOCK_ID,
-        128 /* unused */, OutStreamOptions.defaults(mClientContext));
+        128 /* unused */, OutStreamOptions.defaults(mContext));
 
     // Close stream before closing the writer
     PowerMockito.when(mStream.isCanceled()).thenReturn(true);
diff --git a/core/client/fs/src/test/java/alluxio/client/block/stream/UfsFallbackLocalFileDataWriterTest.java b/core/client/fs/src/test/java/alluxio/client/block/stream/UfsFallbackLocalFileDataWriterTest.java
index 628d6c92f473..7f53b89b9687 100644
--- a/core/client/fs/src/test/java/alluxio/client/block/stream/UfsFallbackLocalFileDataWriterTest.java
+++ b/core/client/fs/src/test/java/alluxio/client/block/stream/UfsFallbackLocalFileDataWriterTest.java
@@ -183,7 +183,7 @@ private DataWriter create(long blockSize, long workerCapacity) throws Exception
     mLocalWriter = new FixedCapacityTestDataWriter(mBuffer);
     DataWriter writer =
         new UfsFallbackLocalFileDataWriter(mLocalWriter, null, mContext, mAddress, BLOCK_ID,
-            blockSize, OutStreamOptions.defaults(mClientContext).setMountId(MOUNT_ID));
+            blockSize, OutStreamOptions.defaults(mContext).setMountId(MOUNT_ID));
     return writer;
   }
 
diff --git a/core/client/fs/src/test/java/alluxio/client/file/AlluxioFileInStreamTest.java b/core/client/fs/src/test/java/alluxio/client/file/AlluxioFileInStreamTest.java
index 1e126fb98392..af828dc9e018 100644
--- a/core/client/fs/src/test/java/alluxio/client/file/AlluxioFileInStreamTest.java
+++ b/core/client/fs/src/test/java/alluxio/client/file/AlluxioFileInStreamTest.java
@@ -186,7 +186,7 @@ public void closeResource() {}
     OpenFilePOptions readOptions =
         OpenFilePOptions.newBuilder().setReadType(ReadPType.CACHE_PROMOTE).build();
     mTestStream = new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, readOptions,
-        mConf), mContext);
+        mConf, mContext), mContext);
   }
 
   @After
@@ -397,7 +397,8 @@ public void longSeekBackwardCachingPartiallyReadBlocks() throws IOException {
     OpenFilePOptions options =
         OpenFilePOptions.newBuilder().setReadType(ReadPType.CACHE_PROMOTE).build();
     mTestStream =
-        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf), mContext);
+        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf, mContext),
+            mContext);
     int seekAmount = (int) (BLOCK_LENGTH / 4 + BLOCK_LENGTH);
     int readAmount = (int) (BLOCK_LENGTH * 3 - BLOCK_LENGTH / 2);
     byte[] buffer = new byte[readAmount];
@@ -420,7 +421,8 @@ public void testSeekWithNoLocalWorker() throws IOException {
     OpenFilePOptions options =
         OpenFilePOptions.newBuilder().setReadType(ReadPType.CACHE_PROMOTE).build();
     mTestStream =
-        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf), mContext);
+        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf, mContext),
+            mContext);
     int readAmount = (int) (BLOCK_LENGTH / 2);
     byte[] buffer = new byte[readAmount];
     // read and seek several times
@@ -439,7 +441,8 @@ public void seekAndClose() throws IOException {
     OpenFilePOptions options =
         OpenFilePOptions.newBuilder().setReadType(ReadPType.CACHE_PROMOTE).build();
     mTestStream =
-        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf), mContext);
+        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf, mContext),
+            mContext);
     int seekAmount = (int) (BLOCK_LENGTH / 2);
     mTestStream.seek(seekAmount);
     mTestStream.close();
@@ -456,7 +459,8 @@ public void shortSeekBackwardCachingPartiallyReadBlocks() throws IOException {
     OpenFilePOptions options =
         OpenFilePOptions.newBuilder().setReadType(ReadPType.CACHE_PROMOTE).build();
     mTestStream =
-        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf), mContext);
+        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf, mContext),
+            mContext);
     int seekAmount = (int) (BLOCK_LENGTH / 4);
     int readAmount = (int) (BLOCK_LENGTH * 2 - BLOCK_LENGTH / 2);
     byte[] buffer = new byte[readAmount];
@@ -482,8 +486,8 @@ public void shortSeekBackwardCachingPartiallyReadBlocks() throws IOException {
   public void longSeekForwardCachingPartiallyReadBlocks() throws IOException {
     OpenFilePOptions options =
         OpenFilePOptions.newBuilder().setReadType(ReadPType.CACHE_PROMOTE).build();
-    mTestStream = new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf),
-        mContext);
+    mTestStream = new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf,
+        mContext), mContext);
     int seekAmount = (int) (BLOCK_LENGTH / 4 + BLOCK_LENGTH);
     int readAmount = (int) (BLOCK_LENGTH / 2);
     byte[] buffer = new byte[readAmount];
@@ -509,7 +513,8 @@ public void shortSeekForwardCachingPartiallyReadBlocks() throws IOException {
     OpenFilePOptions options =
         OpenFilePOptions.newBuilder().setReadType(ReadPType.CACHE_PROMOTE).build();
     mTestStream =
-        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf), mContext);
+        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf, mContext),
+            mContext);
     int seekAmount = (int) (BLOCK_LENGTH / 4);
     int readAmount = (int) (BLOCK_LENGTH * 2 - BLOCK_LENGTH / 2);
     byte[] buffer = new byte[readAmount];
@@ -536,7 +541,8 @@ public void seekBackwardSmallSeekBuffer() throws IOException {
     OpenFilePOptions options =
         OpenFilePOptions.newBuilder().setReadType(ReadPType.CACHE_PROMOTE).build();
     mTestStream =
-        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf), mContext);
+        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf, mContext),
+            mContext);
     int readAmount = (int) (BLOCK_LENGTH / 2);
     byte[] buffer = new byte[readAmount];
     mTestStream.read(buffer);
@@ -555,7 +561,8 @@ public void seekBackwardToFileBeginning() throws IOException {
     OpenFilePOptions options =
         OpenFilePOptions.newBuilder().setReadType(ReadPType.CACHE_PROMOTE).build();
     mTestStream =
-        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf), mContext);
+        new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, options, mConf, mContext),
+            mContext);
     int seekAmount = (int) (BLOCK_LENGTH / 4 + BLOCK_LENGTH);
 
     // Seek forward.
@@ -806,7 +813,7 @@ public void triggerAsyncOnClose() throws Exception {
     OpenFilePOptions readOptions =
         OpenFilePOptions.newBuilder().setReadType(ReadPType.CACHE_PROMOTE).build();
     mTestStream = new AlluxioFileInStream(mStatus, new InStreamOptions(mStatus, readOptions,
-        mConf), mContext);
+        mConf, mContext), mContext);
     mTestStream.read(new byte[(int) mFileSize], 0, (int) mFileSize);
     assertEquals(mFileSize, mTestStream.getPos());
     assertTrue(mTestStream.triggerAsyncCaching(mInStreams.get(mInStreams.size() - 1)));
diff --git a/core/client/fs/src/test/java/alluxio/client/file/FileOutStreamTest.java b/core/client/fs/src/test/java/alluxio/client/file/FileOutStreamTest.java
index 1a0f88cdec9c..a10b1fc3418c 100644
--- a/core/client/fs/src/test/java/alluxio/client/file/FileOutStreamTest.java
+++ b/core/client/fs/src/test/java/alluxio/client/file/FileOutStreamTest.java
@@ -34,12 +34,14 @@
 import alluxio.client.WriteType;
 import alluxio.client.block.BlockStoreClient;
 import alluxio.client.block.BlockWorkerInfo;
+import alluxio.client.block.policy.BlockLocationPolicy;
 import alluxio.client.block.stream.BlockOutStream;
 import alluxio.client.block.stream.TestBlockOutStream;
 import alluxio.client.block.stream.TestUnderFileSystemFileOutStream;
 import alluxio.client.block.stream.UnderFileSystemFileOutStream;
 import alluxio.client.file.options.OutStreamOptions;
 import alluxio.client.util.ClientTestUtils;
+import alluxio.conf.AlluxioConfiguration;
 import alluxio.conf.Configuration;
 import alluxio.conf.InstancedConfiguration;
 import alluxio.conf.PropertyKey;
@@ -119,6 +121,13 @@ public void before() throws Exception {
     when(mFileSystemContext.getClientContext()).thenReturn(mClientContext);
     when(mFileSystemContext.getClusterConf()).thenReturn(sConf);
     when(mFileSystemContext.getPathConf(any(AlluxioURI.class))).thenReturn(sConf);
+    when(mFileSystemContext.getWriteBlockLocationPolicy(any(AlluxioConfiguration.class)))
+        .thenAnswer((Answer) invocation -> {
+          AlluxioConfiguration conf =
+              invocation.getArgument(0, AlluxioConfiguration.class);
+          return BlockLocationPolicy.Factory.create(
+              conf.getClass(PropertyKey.USER_BLOCK_WRITE_LOCATION_POLICY), conf);
+        });
     mBlockStore = PowerMockito.mock(BlockStoreClient.class);
     mFileSystemMasterClient = PowerMockito.mock(FileSystemMasterClient.class);
 
@@ -182,7 +191,7 @@ public void flush() throws IOException {
         mUnderStorageOutputStream);
 
     OutStreamOptions options =
-        OutStreamOptions.defaults(mClientContext).setBlockSizeBytes(BLOCK_LENGTH)
+        OutStreamOptions.defaults(mFileSystemContext).setBlockSizeBytes(BLOCK_LENGTH)
             .setWriteType(WriteType.CACHE_THROUGH).setUfsPath(FILE_NAME.getPath());
     mTestStream = createTestStream(FILE_NAME, options);
   }
@@ -290,7 +299,7 @@ public void flush() throws IOException {
   @Test
   public void cacheWriteExceptionNonSyncPersist() throws IOException {
     OutStreamOptions options =
-        OutStreamOptions.defaults(mClientContext).setBlockSizeBytes(BLOCK_LENGTH)
+        OutStreamOptions.defaults(mFileSystemContext).setBlockSizeBytes(BLOCK_LENGTH)
             .setWriteType(WriteType.MUST_CACHE);
     BlockOutStream stream = mock(BlockOutStream.class);
     when(mBlockStore.getOutStream(anyLong(), anyLong(), any(OutStreamOptions.class)))
@@ -387,7 +396,7 @@ public void writeNullBufferOffset() throws IOException {
   @Test
   public void asyncWrite() throws Exception {
     OutStreamOptions options =
-        OutStreamOptions.defaults(mClientContext).setBlockSizeBytes(BLOCK_LENGTH)
+        OutStreamOptions.defaults(mFileSystemContext).setBlockSizeBytes(BLOCK_LENGTH)
             .setWriteType(WriteType.ASYNC_THROUGH);
     mTestStream = createTestStream(FILE_NAME, options);
 
@@ -415,7 +424,7 @@ public void asyncWriteOptionPropagation() throws Exception {
     OutStreamOptions options =
         new OutStreamOptions(CreateFilePOptions.newBuilder().setWriteType(WritePType.ASYNC_THROUGH)
             .setBlockSizeBytes(BLOCK_LENGTH).setCommonOptions(commonOptions).build(),
-            mClientContext, sConf);
+            mFileSystemContext, sConf);
 
     // Verify that OutStreamOptions have captured the common options properly.
     assertEquals(options.getCommonOptions(), commonOptions);
@@ -442,7 +451,7 @@ public void asyncWriteOptionPropagation() throws Exception {
   public void getBytesWrittenWithDifferentUnderStorageType() throws IOException {
     for (WriteType type : WriteType.values()) {
       OutStreamOptions options =
-          OutStreamOptions.defaults(mClientContext).setBlockSizeBytes(BLOCK_LENGTH)
+          OutStreamOptions.defaults(mFileSystemContext).setBlockSizeBytes(BLOCK_LENGTH)
               .setWriteType(type).setUfsPath(FILE_NAME.getPath());
       mTestStream = createTestStream(FILE_NAME, options);
       mTestStream.write(BufferUtils.getIncreasingByteArray((int) BLOCK_LENGTH));
@@ -456,7 +465,7 @@ public void createWithNoWorker()  {
     // The default 2 minutes is too long.
     sConf.set(PropertyKey.USER_FILE_WRITE_INIT_MAX_DURATION, "10sec");
     mClientContext = ClientContext.create(sConf);
-    OutStreamOptions options = OutStreamOptions.defaults(mClientContext)
+    OutStreamOptions options = OutStreamOptions.defaults(mFileSystemContext)
         .setLocationPolicy((getWorkerOptions) -> Optional.empty())
         .setWriteType(WriteType.CACHE_THROUGH);
     Exception e = assertThrows(UnavailableException.class,
diff --git a/core/client/fs/src/test/java/alluxio/client/file/options/OutStreamOptionsTest.java b/core/client/fs/src/test/java/alluxio/client/file/options/OutStreamOptionsTest.java
index 09a3998eaf44..65e27b5d43e2 100644
--- a/core/client/fs/src/test/java/alluxio/client/file/options/OutStreamOptionsTest.java
+++ b/core/client/fs/src/test/java/alluxio/client/file/options/OutStreamOptionsTest.java
@@ -23,6 +23,7 @@
 import alluxio.client.block.policy.BlockLocationPolicy;
 import alluxio.client.block.policy.LocalFirstPolicy;
 import alluxio.client.block.policy.RoundRobinPolicy;
+import alluxio.client.file.FileSystemContext;
 import alluxio.conf.Configuration;
 import alluxio.conf.InstancedConfiguration;
 import alluxio.conf.PropertyKey;
@@ -91,7 +92,7 @@ public void defaults() throws IOException {
     subject.getPrincipals().add(new User("test_user"));
     ClientContext clientContext = ClientContext.create(subject, mConf);
 
-    OutStreamOptions options = OutStreamOptions.defaults(clientContext);
+    OutStreamOptions options = OutStreamOptions.defaults(FileSystemContext.create(clientContext));
 
     assertEquals(alluxioType, options.getAlluxioStorageType());
     assertEquals(64 * Constants.MB, options.getBlockSizeBytes());
@@ -128,7 +129,7 @@ public void fields() throws Exception {
     mConf.set(PropertyKey.USER_FILE_CREATE_TTL_ACTION, ttlAction);
 
     ClientContext clientContext = ClientContext.create(mConf);
-    OutStreamOptions options = OutStreamOptions.defaults(clientContext);
+    OutStreamOptions options = OutStreamOptions.defaults(FileSystemContext.create(clientContext));
     options.setBlockSizeBytes(blockSize);
     options.setLocationPolicy(locationPolicy);
     options.setOwner(owner);
@@ -154,8 +155,8 @@ public void equalsTest() {
     ClientContext clientContext = ClientContext.create(mConf);
     new EqualsTester()
         .addEqualityGroup(
-            OutStreamOptions.defaults(clientContext),
-            OutStreamOptions.defaults(clientContext))
+            OutStreamOptions.defaults(FileSystemContext.create(clientContext)),
+            OutStreamOptions.defaults(FileSystemContext.create(clientContext)))
         .testEquals();
   }
 }
diff --git a/core/server/worker/src/test/java/alluxio/worker/block/CacheRequestManagerTest.java b/core/server/worker/src/test/java/alluxio/worker/block/CacheRequestManagerTest.java
index e039a110d50b..f88fd5dcc6c9 100644
--- a/core/server/worker/src/test/java/alluxio/worker/block/CacheRequestManagerTest.java
+++ b/core/server/worker/src/test/java/alluxio/worker/block/CacheRequestManagerTest.java
@@ -130,7 +130,7 @@ public void before() throws IOException {
         .setBlockSizeBytes(CHUNK_SIZE)
         .setFileBlockInfos(Collections.singletonList(new FileBlockInfo().setBlockInfo(info))));
     OpenFilePOptions readOptions = FileSystemOptions.openFileDefaults(mConf);
-    InStreamOptions options = new InStreamOptions(dummyStatus, readOptions, mConf);
+    InStreamOptions options = new InStreamOptions(dummyStatus, readOptions, mConf, context);
     mOpenUfsBlockOptions = options.getOpenUfsBlockOptions(BLOCK_ID);
   }
 
diff --git a/core/server/worker/src/test/java/alluxio/worker/block/stream/BlockWorkerDataReaderTest.java b/core/server/worker/src/test/java/alluxio/worker/block/stream/BlockWorkerDataReaderTest.java
index c0ac21540c63..4cf3f8263be4 100644
--- a/core/server/worker/src/test/java/alluxio/worker/block/stream/BlockWorkerDataReaderTest.java
+++ b/core/server/worker/src/test/java/alluxio/worker/block/stream/BlockWorkerDataReaderTest.java
@@ -26,6 +26,7 @@
 import alluxio.Sessions;
 import alluxio.client.block.stream.BlockWorkerDataReader;
 import alluxio.client.block.stream.DataReader;
+import alluxio.client.file.FileSystemContext;
 import alluxio.client.file.URIStatus;
 import alluxio.client.file.options.InStreamOptions;
 import alluxio.conf.Configuration;
@@ -122,7 +123,8 @@ public void before() throws Exception {
     URIStatus dummyStatus =
         new URIStatus(new FileInfo().setBlockIds(Collections.singletonList(BLOCK_ID)));
     InStreamOptions options =
-        new InStreamOptions(dummyStatus, FileSystemOptions.openFileDefaults(mConf), mConf);
+        new InStreamOptions(dummyStatus, FileSystemOptions.openFileDefaults(mConf), mConf,
+            FileSystemContext.create(mConf));
     mDataReaderFactory =
         new BlockWorkerDataReader.Factory(mBlockWorker, BLOCK_ID, CHUNK_SIZE, options);
   }
@@ -151,7 +153,7 @@ public void createAndCloseManyReader() throws Exception {
       mBlockWorker.commitBlock(SESSION_ID, blockId, true);
       InStreamOptions inStreamOptions = new InStreamOptions(
           new URIStatus(new FileInfo().setBlockIds(Collections.singletonList(blockId))),
-          FileSystemOptions.openFileDefaults(mConf), mConf);
+          FileSystemOptions.openFileDefaults(mConf), mConf, FileSystemContext.create(mConf));
       mDataReaderFactory =
           new BlockWorkerDataReader.Factory(mBlockWorker, blockId, CHUNK_SIZE, inStreamOptions);
       DataReader dataReader = mDataReaderFactory.create(0, 100);
@@ -175,7 +177,8 @@ public void readChunkUfs() throws Exception {
         .setFileBlockInfos(Collections.singletonList(new FileBlockInfo().setBlockInfo(info))));
     OpenFilePOptions readOptions = OpenFilePOptions.newBuilder()
         .setReadType(ReadPType.NO_CACHE).build();
-    InStreamOptions options = new InStreamOptions(dummyStatus, readOptions, mConf);
+    InStreamOptions options = new InStreamOptions(dummyStatus, readOptions, mConf,
+        FileSystemContext.create(mConf));
 
     BlockWorkerDataReader.Factory factory = new BlockWorkerDataReader
         .Factory(mBlockWorker, BLOCK_ID, CHUNK_SIZE, options);
diff --git a/tests/src/test/java/alluxio/client/fs/LocalFirstPolicyIntegrationTest.java b/tests/src/test/java/alluxio/client/fs/LocalFirstPolicyIntegrationTest.java
index 41c9d1c90f4b..993bfe1a1b0f 100644
--- a/tests/src/test/java/alluxio/client/fs/LocalFirstPolicyIntegrationTest.java
+++ b/tests/src/test/java/alluxio/client/fs/LocalFirstPolicyIntegrationTest.java
@@ -73,7 +73,7 @@ public void after() {
   }
 
   @Test
-  public void test() throws Exception {
+  public void localNodeFirst() throws Exception {
     AlluxioMasterProcess master = AlluxioMasterProcess.Factory.create();
     WorkerProcess worker1 = AlluxioWorkerProcess.Factory
         .create(TieredIdentityFactory.fromString("node=node1,rack=rack1",
@@ -108,6 +108,30 @@ public void test() throws Exception {
       assertEquals(0, blockWorker2.getStoreMeta().getUsedBytes());
     }
 
+    worker2.stop();
+    worker1.stop();
+    master.stop();
+  }
+
+  @Test
+  public void differentNodeOfSameRack() throws Exception {
+    AlluxioMasterProcess master = AlluxioMasterProcess.Factory.create();
+    WorkerProcess worker1 = AlluxioWorkerProcess.Factory
+        .create(TieredIdentityFactory.fromString("node=node1,rack=rack1",
+            Configuration.global()));
+    WorkerProcess worker2 = AlluxioWorkerProcess.Factory
+        .create(TieredIdentityFactory.fromString("node=node2,rack=rack2",
+            Configuration.global()));
+
+    runProcess(mExecutor, master);
+    runProcess(mExecutor, worker1);
+    runProcess(mExecutor, worker2);
+
+    TestUtils.waitForReady(master);
+    TestUtils.waitForReady(worker1);
+    TestUtils.waitForReady(worker2);
+
+    FileSystem fs = FileSystem.Factory.create();
     // Write to the worker in rack2
     {
       Whitebox.setInternalState(TieredIdentityFactory.class, "sInstance",
@@ -120,9 +144,13 @@ public void test() throws Exception {
       }
       BlockWorker blockWorker1 = worker1.getWorker(BlockWorker.class);
       BlockWorker blockWorker2 = worker2.getWorker(BlockWorker.class);
-      assertEquals(100, blockWorker1.getStoreMeta().getUsedBytes());
+      assertEquals(0, blockWorker1.getStoreMeta().getUsedBytes());
       assertEquals(10, blockWorker2.getStoreMeta().getUsedBytes());
     }
+
+    worker2.stop();
+    worker1.stop();
+    master.stop();
   }
 
   private void runProcess(ExecutorService e, Process p) {
diff --git a/tests/src/test/java/alluxio/server/ft/MultiWorkerIntegrationTest.java b/tests/src/test/java/alluxio/server/ft/MultiWorkerIntegrationTest.java
index 22127468fc54..b5b07b6fb425 100644
--- a/tests/src/test/java/alluxio/server/ft/MultiWorkerIntegrationTest.java
+++ b/tests/src/test/java/alluxio/server/ft/MultiWorkerIntegrationTest.java
@@ -210,7 +210,7 @@ private void replicateFileBlocks(AlluxioURI filePath) throws Exception {
           .orElseThrow(() -> new IllegalStateException("Expected worker"))
           .getNetAddress();
       try (OutputStream outStream = store.getOutStream(blockInfo.getBlockId(),
-          blockInfo.getLength(), dest, OutStreamOptions.defaults(fsContext.getClientContext())
+          blockInfo.getLength(), dest, OutStreamOptions.defaults(fsContext)
               .setBlockSizeBytes(8 * Constants.MB).setWriteType(WriteType.MUST_CACHE))) {
         try (InputStream inStream = store.getInStream(blockInfo.getBlockId(),
             new InStreamOptions(status, Configuration.global()))) {