RATIS-2571. DataStreamInput#readAsync returns ByteBuffers

[amaliujia]

What changes were proposed in this pull request?

In https://github.com/apache/ratis/pull/1488/changes#diff-1164c13fc880c0b59a0752eaa11d0e063f366d107798ab9abfd75d0688631910R55-R61, In order to access the buffer, we realize that the existing DataStreamInput#readAsync would require the end user to know the implementation details.

    if (reply instanceof DataStreamReplyByteBuf) {
      return ((DataStreamReplyByteBuf) reply).slice().nioBuffer();
    } else if (reply instanceof DataStreamReplyByteBuffer) {
      return ((DataStreamReplyByteBuffer) reply).slice();
    }

This PR proposes that DataStreamInput#readAsync returns the Buffers to make it easy to use.

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/RATIS-2571

How was this patch tested?

Existing unit tests

[amaliujia]

@peterxcli @szetszwo

[peterxcli]

LGTM +1

[szetszwo]

@amaliujia , thanks for working on this!

It is a good idea to have methods to get ByteBuffer. Let's add it to DataStreamPacket instead of having a new interface:

diff --git a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamPacketByteBuf.java b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamPacketByteBuf.java
index e6ceeb93d..a6ac94130 100644
--- a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamPacketByteBuf.java
+++ b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamPacketByteBuf.java
@@ -23,6 +23,7 @@ import org.apache.ratis.protocol.ClientId;
 import org.apache.ratis.thirdparty.io.netty.buffer.ByteBuf;
 import org.apache.ratis.thirdparty.io.netty.buffer.Unpooled;
 
+import java.nio.ByteBuffer;
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
@@ -56,6 +57,16 @@ public class DataStreamPacketByteBuf extends DataStreamPacketImpl {
     return getBuf().slice();
   }
 
+  @Override
+  public ByteBuffer nioBuffer() {
+    return getBuf().nioBuffer();
+  }
+
+  @Override
+  public ByteBuffer[] nioBuffers() {
+    return getBuf().nioBuffers();
+  }
+
   public final void release() {
     final ByteBuf got = buf.getAndSet(null);
     if (got != null) {
diff --git a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamPacketByteBuffer.java b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamPacketByteBuffer.java
index b8d7b48a7..a7d6f49d9 100644
--- a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamPacketByteBuffer.java
+++ b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamPacketByteBuffer.java
@@ -44,4 +44,14 @@ public abstract class DataStreamPacketByteBuffer extends DataStreamPacketImpl {
   public ByteBuffer slice() {
     return buffer.slice();
   }
+
+  @Override
+  public ByteBuffer nioBuffer() {
+    return slice();
+  }
+
+  @Override
+  public ByteBuffer[] nioBuffers() {
+    return new ByteBuffer[]{slice()};
+  }
 }
diff --git a/ratis-common/src/main/java/org/apache/ratis/protocol/DataStreamPacket.java b/ratis-common/src/main/java/org/apache/ratis/protocol/DataStreamPacket.java
index caebb9e9b..95c5e6211 100644
--- a/ratis-common/src/main/java/org/apache/ratis/protocol/DataStreamPacket.java
+++ b/ratis-common/src/main/java/org/apache/ratis/protocol/DataStreamPacket.java
@@ -20,6 +20,8 @@ package org.apache.ratis.protocol;
 
 import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type;
 
+import java.nio.ByteBuffer;
+
 public interface DataStreamPacket {
   ClientId getClientId();
 
@@ -30,4 +32,12 @@ public interface DataStreamPacket {
   long getStreamOffset();
 
   long getDataLength();
+
+  default ByteBuffer nioBuffer() {
+    throw new UnsupportedOperationException();
+  }
+
+  default ByteBuffer[] nioBuffers() {
+    throw new UnsupportedOperationException();
+  }
 }
\ No newline at end of file

[amaliujia]

Done. Indeed this propose is much easier and cleaner

[szetszwo]

+1 the change looks good.

[szetszwo]

@amaliujia , thanks for working on this!

@peterxcli , thanks for reviewing this!