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 a8573749d RATIS-2563. Add DataStreamReplyByteBuf (#1485)
a8573749d is described below
commit a8573749d9ff99a704b5df5ed57cdfed91b4d3e0
Author: Peter Lee <[email protected]>
AuthorDate: Thu Jun 18 00:35:32 2026 +0800
RATIS-2563. Add DataStreamReplyByteBuf (#1485)
---
...stByteBuf.java => DataStreamPacketByteBuf.java} | 41 +++-----
.../datastream/impl/DataStreamReplyBuilder.java | 112 +++++++++++++++++++++
.../datastream/impl/DataStreamReplyByteBuf.java | 99 ++++++++++++++++++
.../datastream/impl/DataStreamReplyByteBuffer.java | 78 +++-----------
.../datastream/impl/DataStreamRequestByteBuf.java | 35 +------
5 files changed, 240 insertions(+), 125 deletions(-)
diff --git
a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamRequestByteBuf.java
b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamPacketByteBuf.java
similarity index 54%
copy from
ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamRequestByteBuf.java
copy to
ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamPacketByteBuf.java
index 1873bec9b..76400ff80 100644
---
a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamRequestByteBuf.java
+++
b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamPacketByteBuf.java
@@ -18,64 +18,47 @@
package org.apache.ratis.datastream.impl;
-import org.apache.ratis.io.WriteOption;
import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type;
import org.apache.ratis.protocol.ClientId;
-import org.apache.ratis.protocol.DataStreamRequest;
-import org.apache.ratis.protocol.DataStreamRequestHeader;
-import org.apache.ratis.thirdparty.com.google.common.collect.Lists;
import org.apache.ratis.thirdparty.io.netty.buffer.ByteBuf;
import org.apache.ratis.thirdparty.io.netty.buffer.Unpooled;
-import java.util.Collections;
-import java.util.List;
-import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
/**
- * Implements {@link DataStreamRequest} with {@link ByteBuf}.
+ * Extends {@link DataStreamPacketImpl} with {@link ByteBuf}.
* <p>
* This class is immutable.
*/
-public class DataStreamRequestByteBuf extends DataStreamPacketImpl implements
DataStreamRequest {
+class DataStreamPacketByteBuf extends DataStreamPacketImpl {
private final AtomicReference<ByteBuf> buf;
- private final List<WriteOption> options;
- public DataStreamRequestByteBuf(ClientId clientId, Type type, long streamId,
long streamOffset,
- Iterable<WriteOption> options, ByteBuf buf) {
+ DataStreamPacketByteBuf(ClientId clientId, Type type, long streamId, long
streamOffset, ByteBuf buf) {
super(clientId, type, streamId, streamOffset);
this.buf = new AtomicReference<>(buf != null? buf.asReadOnly():
Unpooled.EMPTY_BUFFER);
- this.options = Collections.unmodifiableList(Lists.newArrayList(options));
}
- public DataStreamRequestByteBuf(DataStreamRequestHeader header, ByteBuf buf)
{
- this(header.getClientId(), header.getType(), header.getStreamId(),
header.getStreamOffset(),
- header.getWriteOptionList(), buf);
- }
-
- ByteBuf getBuf() {
- return Optional.ofNullable(buf.get()).orElseThrow(
- () -> new IllegalStateException("buf is already released in " + this));
+ final ByteBuf getBuf() {
+ final ByteBuf got = buf.get();
+ if (got == null) {
+ throw new IllegalStateException("buf is already released in " + this);
+ }
+ return got;
}
@Override
- public long getDataLength() {
+ public final long getDataLength() {
return getBuf().readableBytes();
}
- public ByteBuf slice() {
+ public final ByteBuf slice() {
return getBuf().slice();
}
- public void release() {
+ public final void release() {
final ByteBuf got = buf.getAndSet(null);
if (got != null) {
got.release();
}
}
-
- @Override
- public List<WriteOption> getWriteOptionList() {
- return options;
- }
}
diff --git
a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyBuilder.java
b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyBuilder.java
new file mode 100644
index 000000000..3c99bab92
--- /dev/null
+++
b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyBuilder.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ratis.datastream.impl;
+
+import org.apache.ratis.proto.RaftProtos;
+import org.apache.ratis.protocol.ClientId;
+import org.apache.ratis.protocol.DataStreamPacket;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+
+@SuppressWarnings("checkstyle:HiddenField")
+abstract class DataStreamReplyBuilder<B extends DataStreamReplyBuilder<B>> {
+ private ClientId clientId;
+ private RaftProtos.DataStreamPacketHeaderProto.Type type;
+ private long streamId;
+ private long streamOffset;
+
+ private boolean success;
+ private long bytesWritten;
+ private Collection<RaftProtos.CommitInfoProto> commitInfos =
Collections.emptyList();
+
+ abstract B getThis();
+
+ public final ClientId getClientId() {
+ return clientId;
+ }
+
+ public final B setClientId(ClientId clientId) {
+ this.clientId = clientId;
+ return getThis();
+ }
+
+ public final RaftProtos.DataStreamPacketHeaderProto.Type getType() {
+ return type;
+ }
+
+ public final B setType(RaftProtos.DataStreamPacketHeaderProto.Type type) {
+ this.type = type;
+ return getThis();
+ }
+
+ public final long getStreamId() {
+ return streamId;
+ }
+
+ public final B setStreamId(long streamId) {
+ this.streamId = streamId;
+ return getThis();
+ }
+
+ public final long getStreamOffset() {
+ return streamOffset;
+ }
+
+ public final B setStreamOffset(long streamOffset) {
+ this.streamOffset = streamOffset;
+ return getThis();
+ }
+
+ public final boolean isSuccess() {
+ return success;
+ }
+
+ public final B setSuccess(boolean success) {
+ this.success = success;
+ return getThis();
+ }
+
+ public final long getBytesWritten() {
+ return bytesWritten;
+ }
+
+ public final B setBytesWritten(long bytesWritten) {
+ this.bytesWritten = bytesWritten;
+ return getThis();
+ }
+
+ public final Collection<RaftProtos.CommitInfoProto> getCommitInfos() {
+ return commitInfos;
+ }
+
+ public final B setCommitInfos(Collection<RaftProtos.CommitInfoProto>
commitInfos) {
+ this.commitInfos = commitInfos != null
+ ? Collections.unmodifiableCollection(new ArrayList<>(commitInfos))
+ : Collections.emptyList();
+ return getThis();
+ }
+
+ public final B setDataStreamPacket(DataStreamPacket packet) {
+ return setClientId(packet.getClientId())
+ .setType(packet.getType())
+ .setStreamId(packet.getStreamId())
+ .setStreamOffset(packet.getStreamOffset());
+ }
+}
diff --git
a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyByteBuf.java
b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyByteBuf.java
new file mode 100644
index 000000000..bef58563f
--- /dev/null
+++
b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyByteBuf.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ratis.datastream.impl;
+
+import org.apache.ratis.proto.RaftProtos.CommitInfoProto;
+import org.apache.ratis.protocol.DataStreamReply;
+import org.apache.ratis.protocol.DataStreamReplyHeader;
+import org.apache.ratis.thirdparty.io.netty.buffer.ByteBuf;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * Implements {@link DataStreamReply} with {@link ByteBuf}.
+ * <p>
+ * This class is immutable.
+ */
+public final class DataStreamReplyByteBuf extends DataStreamPacketByteBuf
implements DataStreamReply {
+ public static final class Builder extends DataStreamReplyBuilder<Builder> {
+ private ByteBuf buf;
+
+ private Builder() {
+ }
+
+ @Override
+ Builder getThis() {
+ return this;
+ }
+
+ public Builder setBuf(ByteBuf newBuf) {
+ this.buf = newBuf != null ? newBuf.asReadOnly() : null;
+ return this;
+ }
+
+ public Builder setDataStreamReplyHeader(DataStreamReplyHeader header) {
+ return setDataStreamPacket(header)
+ .setSuccess(header.isSuccess())
+ .setBytesWritten(header.getBytesWritten())
+ .setCommitInfos(header.getCommitInfos());
+ }
+
+ public DataStreamReplyByteBuf build() {
+ return new DataStreamReplyByteBuf(this);
+ }
+ }
+
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ private final boolean success;
+ private final long bytesWritten;
+ private final Collection<CommitInfoProto> commitInfos;
+
+ private DataStreamReplyByteBuf(DataStreamReplyByteBuf.Builder b) {
+ super(b.getClientId(), b.getType(), b.getStreamId(), b.getStreamOffset(),
b.buf);
+
+ this.success = b.isSuccess();
+ this.bytesWritten = b.getBytesWritten();
+ this.commitInfos = b.getCommitInfos();
+ }
+
+ @Override
+ public boolean isSuccess() {
+ return success;
+ }
+
+ @Override
+ public long getBytesWritten() {
+ return bytesWritten;
+ }
+
+ @Override
+ public Collection<CommitInfoProto> getCommitInfos() {
+ return Collections.unmodifiableCollection(commitInfos);
+ }
+
+ @Override
+ public String toString() {
+ return super.toString()
+ + "," + (success ? "SUCCESS" : "FAILED")
+ + ",bytesWritten=" + bytesWritten;
+ }
+}
diff --git
a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyByteBuffer.java
b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyByteBuffer.java
index 5cb5569be..7138a61a2 100644
---
a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyByteBuffer.java
+++
b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyByteBuffer.java
@@ -18,72 +18,25 @@
package org.apache.ratis.datastream.impl;
import org.apache.ratis.proto.RaftProtos.CommitInfoProto;
-import org.apache.ratis.protocol.ClientId;
-import org.apache.ratis.protocol.DataStreamPacket;
import org.apache.ratis.protocol.DataStreamReply;
import org.apache.ratis.protocol.DataStreamReplyHeader;
-import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type;
import java.nio.ByteBuffer;
import java.util.Collection;
-import java.util.Collections;
/**
* Implements {@link DataStreamReply} with {@link ByteBuffer}.
- *
+ * <p>
* This class is immutable.
*/
public final class DataStreamReplyByteBuffer extends
DataStreamPacketByteBuffer implements DataStreamReply {
- public static final class Builder {
- private ClientId clientId;
- private Type type;
- private long streamId;
- private long streamOffset;
+ public static final class Builder extends DataStreamReplyBuilder<Builder> {
private ByteBuffer buffer;
- private boolean success;
- private long bytesWritten;
- private Collection<CommitInfoProto> commitInfos;
-
private Builder() {}
- public Builder setClientId(ClientId clientId) {
- this.clientId = clientId;
- return this;
- }
-
- public Builder setType(Type type) {
- this.type = type;
- return this;
- }
-
- public Builder setStreamId(long streamId) {
- this.streamId = streamId;
- return this;
- }
-
- public Builder setStreamOffset(long streamOffset) {
- this.streamOffset = streamOffset;
- return this;
- }
-
- public Builder setBuffer(ByteBuffer buffer) {
- this.buffer = buffer;
- return this;
- }
-
- public Builder setSuccess(boolean success) {
- this.success = success;
- return this;
- }
-
- public Builder setBytesWritten(long bytesWritten) {
- this.bytesWritten = bytesWritten;
- return this;
- }
-
- public Builder setCommitInfos(Collection<CommitInfoProto> commitInfos) {
- this.commitInfos = commitInfos;
+ @Override
+ Builder getThis() {
return this;
}
@@ -94,16 +47,13 @@ public final class DataStreamReplyByteBuffer extends
DataStreamPacketByteBuffer
.setCommitInfos(header.getCommitInfos());
}
- public Builder setDataStreamPacket(DataStreamPacket packet) {
- return setClientId(packet.getClientId())
- .setType(packet.getType())
- .setStreamId(packet.getStreamId())
- .setStreamOffset(packet.getStreamOffset());
+ public Builder setBuffer(ByteBuffer buffer) {
+ this.buffer = buffer;
+ return getThis();
}
public DataStreamReplyByteBuffer build() {
- return new DataStreamReplyByteBuffer(
- clientId, type, streamId, streamOffset, buffer, success,
bytesWritten, commitInfos);
+ return new DataStreamReplyByteBuffer(this);
}
}
@@ -115,14 +65,12 @@ public final class DataStreamReplyByteBuffer extends
DataStreamPacketByteBuffer
private final long bytesWritten;
private final Collection<CommitInfoProto> commitInfos;
- @SuppressWarnings("parameternumber")
- private DataStreamReplyByteBuffer(ClientId clientId, Type type, long
streamId, long streamOffset, ByteBuffer buffer,
- boolean success, long bytesWritten, Collection<CommitInfoProto>
commitInfos) {
- super(clientId, type, streamId, streamOffset, buffer);
+ private DataStreamReplyByteBuffer(Builder b) {
+ super(b.getClientId(), b.getType(), b.getStreamId(), b.getStreamOffset(),
b.buffer);
- this.success = success;
- this.bytesWritten = bytesWritten;
- this.commitInfos = commitInfos != null? commitInfos:
Collections.emptyList();
+ this.success = b.isSuccess();
+ this.bytesWritten = b.getBytesWritten();
+ this.commitInfos = b.getCommitInfos();
}
@Override
diff --git
a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamRequestByteBuf.java
b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamRequestByteBuf.java
index 1873bec9b..a88a81f68 100644
---
a/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamRequestByteBuf.java
+++
b/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamRequestByteBuf.java
@@ -23,29 +23,23 @@ import
org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type;
import org.apache.ratis.protocol.ClientId;
import org.apache.ratis.protocol.DataStreamRequest;
import org.apache.ratis.protocol.DataStreamRequestHeader;
-import org.apache.ratis.thirdparty.com.google.common.collect.Lists;
import org.apache.ratis.thirdparty.io.netty.buffer.ByteBuf;
-import org.apache.ratis.thirdparty.io.netty.buffer.Unpooled;
import java.util.Collections;
import java.util.List;
-import java.util.Optional;
-import java.util.concurrent.atomic.AtomicReference;
/**
* Implements {@link DataStreamRequest} with {@link ByteBuf}.
* <p>
* This class is immutable.
*/
-public class DataStreamRequestByteBuf extends DataStreamPacketImpl implements
DataStreamRequest {
- private final AtomicReference<ByteBuf> buf;
+public class DataStreamRequestByteBuf extends DataStreamPacketByteBuf
implements DataStreamRequest {
private final List<WriteOption> options;
public DataStreamRequestByteBuf(ClientId clientId, Type type, long streamId,
long streamOffset,
- Iterable<WriteOption> options, ByteBuf buf) {
- super(clientId, type, streamId, streamOffset);
- this.buf = new AtomicReference<>(buf != null? buf.asReadOnly():
Unpooled.EMPTY_BUFFER);
- this.options = Collections.unmodifiableList(Lists.newArrayList(options));
+ List<WriteOption> options, ByteBuf buf) {
+ super(clientId, type, streamId, streamOffset, buf);
+ this.options = Collections.unmodifiableList(options);
}
public DataStreamRequestByteBuf(DataStreamRequestHeader header, ByteBuf buf)
{
@@ -53,27 +47,6 @@ public class DataStreamRequestByteBuf extends
DataStreamPacketImpl implements Da
header.getWriteOptionList(), buf);
}
- ByteBuf getBuf() {
- return Optional.ofNullable(buf.get()).orElseThrow(
- () -> new IllegalStateException("buf is already released in " + this));
- }
-
- @Override
- public long getDataLength() {
- return getBuf().readableBytes();
- }
-
- public ByteBuf slice() {
- return getBuf().slice();
- }
-
- public void release() {
- final ByteBuf got = buf.getAndSet(null);
- if (got != null) {
- got.release();
- }
- }
-
@Override
public List<WriteOption> getWriteOptionList() {
return options;
