(ratis) branch master updated: RATIS-2571. DataStreamInput#readAsync returns ByteBuffers (#1492)

[szetszwo]

This is an automated email from the ASF dual-hosted git repository.

szetszwo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ratis.git


The following commit(s) were added to refs/heads/master by this push:
     new 898a39cb7 RATIS-2571. DataStreamInput#readAsync returns ByteBuffers 
(#1492)
898a39cb7 is described below

commit 898a39cb760ed44ee5945da2a23a77a5792c2b4a
Author: Rui Wang <amaliu...@apache.org>
AuthorDate: Sat Jun 27 00:39:58 2026 +0800

    RATIS-2571. DataStreamInput#readAsync returns ByteBuffers (#1492)
---
 .../java/org/apache/ratis/client/api/DataStreamInput.java     |  2 ++
 .../apache/ratis/datastream/impl/DataStreamPacketByteBuf.java | 11 +++++++++++
 .../ratis/datastream/impl/DataStreamPacketByteBuffer.java     | 10 ++++++++++
 .../main/java/org/apache/ratis/protocol/DataStreamPacket.java | 10 ++++++++++
 4 files changed, 33 insertions(+)

diff --git 
a/ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamInput.java 
b/ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamInput.java
index e50033405..0eff3cab1 100644
--- 
a/ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamInput.java
+++ 
b/ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamInput.java
@@ -30,6 +30,8 @@ public interface DataStreamInput extends Closeable {
   /**
    * Read the next chunk in the stream asynchronously.
    * The caller owns the returned {@link DataStreamReply} which is a {@link 
ReferenceCountedObject}.
+   * and a {@link DataStreamPacket}. Access the buffer via {@link 
DataStreamPacket#nioBuffer()}
+   * or {@link DataStreamPacket#nioBuffers()}.
    * It must call {@link ReferenceCountedObject#release()} after consuming it.
    *
    * @return a future of the reference-counted reply.
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