Disable short circuit IO for Dora by default

core/client/fs/pom.xml

@@ -98,6 +98,12 @@
       <version>${project.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.alluxio</groupId>
+      <artifactId>alluxio-underfs-local</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

core/client/fs/src/main/java/alluxio/client/block/stream/BlockInStream.java

@@ -118,6 +118,8 @@ public static BlockInStream create(FileSystemContext context, BlockInfo info,
         alluxioConf.getBoolean(PropertyKey.USER_SHORT_CIRCUIT_PREFERRED);
     boolean sourceSupportsDomainSocket = NettyUtils.isDomainSocketSupported(dataSource);
     boolean sourceIsLocal = dataSourceType == BlockInStreamSource.NODE_LOCAL;
+    boolean nettyTransEnabled =
+        alluxioConf.getBoolean(PropertyKey.USER_NETTY_DATA_TRANSMISSION_ENABLED);
 
     // Short circuit is enabled when
     // 1. data source is local node
@@ -136,6 +138,18 @@ public static BlockInStream create(FileSystemContext context, BlockInfo info,
       }
     }
 
+    // use Netty to transfer data
+    if (nettyTransEnabled) {
+      // TODO(JiamingMai): implement this logic
+      LOG.debug("Creating Netty input stream for block {} @ {} from client {} reading through {} ("
+              + "data locates in the local worker {}, shortCircuitEnabled {}, "
+              + "shortCircuitPreferred {}, sourceSupportDomainSocket {})",
+          blockId, dataSource, NetworkAddressUtils.getClientHostName(alluxioConf), dataSource,
+          sourceIsLocal, shortCircuit, shortCircuitPreferred, sourceSupportsDomainSocket);
+      return createNettyBlockInStream(context, dataSource, dataSourceType, blockId,
+          blockSize, options);
+    }
+
     // gRPC
     LOG.debug("Creating gRPC input stream for block {} @ {} from client {} reading through {} ("
         + "data locates in the local worker {}, shortCircuitEnabled {}, "
@@ -190,6 +204,32 @@ private static BlockInStream createLocalBlockInStream(FileSystemContext context,
         address, BlockInStreamSource.NODE_LOCAL, blockId, length);
   }
 
+  /**
+   * Creates a {@link BlockInStream} to read from a Netty data server.
+   *
+   * @param context the file system context
+   * @param address the address of the gRPC data server
+   * @param blockSource the source location of the block
+   * @param blockSize the block size
+   * @param blockId the block id
+   * @return the {@link BlockInStream} created
+   */
+  private static BlockInStream createNettyBlockInStream(FileSystemContext context,
+      WorkerNetAddress address, BlockInStreamSource blockSource,
+      long blockId, long blockSize, InStreamOptions options) {
+    AlluxioConfiguration conf = context.getClusterConf();
+    long chunkSize = conf.getBytes(
+        PropertyKey.USER_STREAMING_READER_CHUNK_SIZE_BYTES);
+    // Construct the partial read request
+    Protocol.ReadRequest.Builder builder = Protocol.ReadRequest.newBuilder()
+        .setBlockId(blockId)
+        .setPromote(ReadType.fromProto(options.getOptions().getReadType()).isPromote())
+        .setOpenUfsBlockOptions(options.getOpenUfsBlockOptions(blockId)) // Add UFS fallback options
+        .setChunkSize(chunkSize);
+    DataReader.Factory factory = new NettyDataReader.Factory(context, address, builder);
+    return new BlockInStream(factory, address, blockSource, blockId, blockSize);
+  }
+
   /**
    * Creates a {@link BlockInStream} to read from a gRPC data server.
    *

core/client/fs/src/main/java/alluxio/client/block/stream/BlockWorkerClient.java

@@ -17,6 +17,8 @@
 import alluxio.grpc.ClearMetricsResponse;
 import alluxio.grpc.CreateLocalBlockRequest;
 import alluxio.grpc.CreateLocalBlockResponse;
+import alluxio.grpc.GetStatusPRequest;
+import alluxio.grpc.GetStatusPResponse;
 import alluxio.grpc.GrpcServerAddress;
 import alluxio.grpc.LoadRequest;
 import alluxio.grpc.LoadResponse;
@@ -43,6 +45,7 @@
  * gRPC client for worker communication.
  */
 public interface BlockWorkerClient extends Closeable {
+
   /**
    * Factory for block worker client.
    */
@@ -161,4 +164,12 @@ StreamObserver<OpenLocalBlockRequest> openLocalBlock(
    * @throws StatusRuntimeException if any error occurs
    */
   ListenableFuture<LoadResponse> load(LoadRequest request);
+
+  /**
+   * get file status.
+   *
+   * @param request
+   * @return status
+   */
+  GetStatusPResponse getStatus(GetStatusPRequest request);
 }

core/client/fs/src/main/java/alluxio/client/block/stream/DataWriter.java

@@ -62,6 +62,8 @@ public static DataWriter create(FileSystemContext context, long blockId, long bl
       boolean ufsFallbackEnabled = options.getWriteType() == WriteType.ASYNC_THROUGH
           && alluxioConf.getBoolean(PropertyKey.USER_FILE_UFS_TIER_ENABLED);
       boolean workerIsLocal = CommonUtils.isLocalHost(address, alluxioConf);
+      boolean nettyTransEnabled =
+          alluxioConf.getBoolean(PropertyKey.USER_NETTY_DATA_TRANSMISSION_ENABLED);
 
       if (workerIsLocal && context.hasProcessLocalWorker() && !ufsFallbackEnabled) {
         LOG.debug("Creating worker process local output stream for block {} @ {}",
@@ -84,16 +86,28 @@ public static DataWriter create(FileSystemContext context, long blockId, long bl
         }
         LOG.debug("Creating short circuit output stream for block {} @ {}", blockId, address);
         return LocalFileDataWriter.create(context, address, blockId, blockSize, options);
-      } else {
-        LOG.debug("Creating gRPC output stream for block {} @ {} from client {} "
-            + "(data locates in local worker: {}, shortCircuitEnabled: {}, "
-            + "shortCircuitPreferred: {}, domainSocketSupported: {})",
+      }
+
+      if (nettyTransEnabled) {
+        LOG.debug("Creating netty output stream for block {} @ {} from client {} "
+                + "(data locates in local worker: {}, shortCircuitEnabled: {}, "
+                + "shortCircuitPreferred: {}, domainSocketSupported: {})",
             blockId, address, NetworkAddressUtils.getClientHostName(alluxioConf),
             workerIsLocal, shortCircuit, shortCircuitPreferred, domainSocketSupported);
-        return GrpcDataWriter
+        // TODO(JiamingMai): implement the netty writer here
+        return NettyDataWriter
             .create(context, address, blockId, blockSize, RequestType.ALLUXIO_BLOCK,
                 options);
       }
+
+      LOG.debug("Creating gRPC output stream for block {} @ {} from client {} "
+          + "(data locates in local worker: {}, shortCircuitEnabled: {}, "
+          + "shortCircuitPreferred: {}, domainSocketSupported: {})",
+          blockId, address, NetworkAddressUtils.getClientHostName(alluxioConf),
+          workerIsLocal, shortCircuit, shortCircuitPreferred, domainSocketSupported);
+      return GrpcDataWriter
+          .create(context, address, blockId, blockSize, RequestType.ALLUXIO_BLOCK,
+              options);
     }
   }
 

core/client/fs/src/main/java/alluxio/client/block/stream/DefaultBlockWorkerClient.java

@@ -24,6 +24,8 @@
 import alluxio.grpc.DataMessageMarshaller;
 import alluxio.grpc.DataMessageMarshallerProvider;
 import alluxio.grpc.FreeWorkerRequest;
+import alluxio.grpc.GetStatusPRequest;
+import alluxio.grpc.GetStatusPResponse;
 import alluxio.grpc.GrpcChannel;
 import alluxio.grpc.GrpcChannelBuilder;
 import alluxio.grpc.GrpcNetworkGroup;
@@ -248,4 +250,10 @@ public void freeWorker() {
   public ListenableFuture<LoadResponse> load(LoadRequest request) {
     return mRpcFutureStub.load(request);
   }
+
+  @Override
+  public GetStatusPResponse getStatus(GetStatusPRequest request) {
+    return mRpcBlockingStub.withDeadlineAfter(mRpcTimeoutMs, TimeUnit.MILLISECONDS)
+        .getStatus(request);
+  }
 }