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 0b0af3576 RATIS-2568. Add Streaming Read to FileStore (#1488)
0b0af3576 is described below
commit 0b0af35762f5da609d47f5094a7c667991d87e99
Author: Rui Wang <[email protected]>
AuthorDate: Thu Jul 9 03:10:47 2026 +0800
RATIS-2568. Add Streaming Read to FileStore (#1488)
---
.../apache/ratis/examples/filestore/FileInfo.java | 36 ++++-
.../apache/ratis/examples/filestore/FileStore.java | 21 ++-
.../ratis/examples/filestore/FileStoreClient.java | 48 +++++++
.../examples/filestore/FileStoreStateMachine.java | 19 +++
.../ratis/examples/filestore/cli/FileStore.java | 1 +
.../apache/ratis/examples/filestore/cli/Read.java | 158 +++++++++++++++++++++
.../filestore/FileStoreStreamingBaseTest.java | 36 +++++
.../ratis/examples/filestore/FileStoreWriter.java | 36 ++++-
8 files changed, 345 insertions(+), 10 deletions(-)
diff --git
a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileInfo.java
b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileInfo.java
index c3ec21f64..bdb099a60 100644
---
a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileInfo.java
+++
b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileInfo.java
@@ -19,11 +19,7 @@ package org.apache.ratis.examples.filestore;
import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
-import org.apache.ratis.util.CollectionUtils;
-import org.apache.ratis.util.JavaUtils;
-import org.apache.ratis.util.LogUtils;
-import org.apache.ratis.util.Preconditions;
-import org.apache.ratis.util.TaskQueue;
+import org.apache.ratis.util.*;
import org.apache.ratis.util.function.CheckedFunction;
import org.apache.ratis.util.function.CheckedSupplier;
import org.slf4j.Logger;
@@ -31,7 +27,9 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
import java.nio.channels.SeekableByteChannel;
+import java.nio.channels.WritableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
@@ -88,6 +86,28 @@ abstract class FileInfo {
}
}
+ void streamRead(CheckedFunction<Path, Path, IOException> resolver, long
offset, long length,
+ WritableByteChannel stream) throws IOException {
+ if (offset + length > getWriteSize()) {
+ throw new IOException("Failed to read: offset (=" + offset
+ + " + length (=" + length + ") > size = " + getWriteSize()
+ + ", path=" + getRelativePath());
+ }
+
+ try (FileChannel in = FileUtils.newFileChannel(
+ resolver.apply(getRelativePath()), StandardOpenOption.READ)) {
+ long transferred = 0;
+ while (transferred < length) {
+ final long n = in.transferTo(offset + transferred, length -
transferred, stream);
+ Preconditions.assertTrue(n >= 0);
+ transferred += n;
+ }
+ Preconditions.assertSame(length, transferred, "transferred");
+ } finally {
+ stream.close();
+ }
+ }
+
UnderConstruction asUnderConstruction() {
throw new UnsupportedOperationException(
"File " + getRelativePath() + " is not under construction.");
@@ -121,6 +141,12 @@ abstract class FileInfo {
this.writeSize = f.getWriteSize();
}
+ ReadOnly(Path relativePath, long size) {
+ super(relativePath);
+ this.committedSize = size;
+ this.writeSize = size;
+ }
+
@Override
long getCommittedSize() {
return committedSize;
diff --git
a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStore.java
b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStore.java
index a930170ec..585660e1e 100644
---
a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStore.java
+++
b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStore.java
@@ -43,6 +43,7 @@ import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
+import java.nio.channels.WritableByteChannel;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
@@ -102,6 +103,11 @@ public class FileStore implements Closeable {
}
}
+ void putReadOnly(ReadOnly ro) {
+ LOG.trace("{}: putReadOnly {}", name, ro.getRelativePath());
+ map.put(ro.getRelativePath(), ro);
+ }
+
ReadOnly close(UnderConstruction uc) {
LOG.trace("{}: close {}", name, uc.getRelativePath());
final ReadOnly ro = new ReadOnly(uc);
@@ -208,6 +214,12 @@ public class FileStore implements Closeable {
return submit(task, reader);
}
+ void streamRead(String relative, long offset, long length,
WritableByteChannel stream)
+ throws IOException {
+ final FileInfo info = files.get(relative);
+ info.streamRead(this::resolve, offset, length, stream);
+ }
+
CompletableFuture<Path> delete(long index, String relative) {
final Supplier<String> name = () -> "delete(" + relative + ") @" + getId()
+ ":" + index;
final CheckedSupplier<Path, IOException> task =
LogUtils.newCheckedSupplier(LOG, () -> {
@@ -284,13 +296,18 @@ public class FileStore implements Closeable {
CompletableFuture<StreamWriteReplyProto> streamCommit(String p, long
bytesWritten) {
return CompletableFuture.supplyAsync(() -> {
+ final Path relative = normalize(p);
final long len;
- try (RandomAccessFile file = new
RandomAccessFile(resolve(normalize(p)).toFile(), "r")) {
+ try (RandomAccessFile file = new
RandomAccessFile(resolve(relative).toFile(), "r")) {
len = file.length();
- return StreamWriteReplyProto.newBuilder().setIsSuccess(len ==
bytesWritten).setByteWritten(len).build();
} catch (IOException e) {
throw new CompletionException("Failed to commit stream " + p + " with
" + bytesWritten + " B.", e);
}
+ final boolean success = len == bytesWritten;
+ if (success) {
+ files.putReadOnly(new ReadOnly(relative, len));
+ }
+ return
StreamWriteReplyProto.newBuilder().setIsSuccess(success).setByteWritten(len).build();
}, committer);
}
diff --git
a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreClient.java
b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreClient.java
index fb0ee49dc..8eb6e3c17 100644
---
a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreClient.java
+++
b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreClient.java
@@ -18,6 +18,7 @@
package org.apache.ratis.examples.filestore;
import org.apache.ratis.client.RaftClient;
+import org.apache.ratis.client.api.DataStreamInput;
import org.apache.ratis.client.api.DataStreamOutput;
import org.apache.ratis.conf.RaftProperties;
import org.apache.ratis.proto.ExamplesProtos.DeleteReplyProto;
@@ -29,6 +30,8 @@ import
org.apache.ratis.proto.ExamplesProtos.StreamWriteRequestProto;
import org.apache.ratis.proto.ExamplesProtos.WriteReplyProto;
import org.apache.ratis.proto.ExamplesProtos.WriteRequestHeaderProto;
import org.apache.ratis.proto.ExamplesProtos.WriteRequestProto;
+import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type;
+import org.apache.ratis.protocol.DataStreamReply;
import org.apache.ratis.protocol.Message;
import org.apache.ratis.protocol.RaftClientReply;
import org.apache.ratis.protocol.RaftGroup;
@@ -39,6 +42,7 @@ import
org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
import org.apache.ratis.util.JavaUtils;
import org.apache.ratis.util.Preconditions;
import org.apache.ratis.util.ProtoUtils;
+import org.apache.ratis.util.ReferenceCountedObject;
import org.apache.ratis.util.function.CheckedFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,6 +50,7 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.channels.WritableByteChannel;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.function.Function;
@@ -197,6 +202,49 @@ public class FileStoreClient implements Closeable {
return
client.getDataStreamApi().stream(request.toByteString().asReadOnlyByteBuffer(),
routingTable);
}
+ public DataStreamInput getStreamInput(String path, long offset, long length)
{
+ final ReadRequestProto read = ReadRequestProto.newBuilder()
+ .setPath(ProtoUtils.toByteString(path))
+ .setOffset(offset)
+ .setLength(length)
+ .build();
+ return
client.getDataStreamApi().streamReadOnly(read.toByteString().asReadOnlyByteBuffer());
+ }
+
+ /**
+ * Read file data using streaming read and write it to the given channel.
+ *
+ * @return total number of bytes read.
+ */
+ public long streamRead(String path, long offset, long length,
WritableByteChannel channel)
+ throws IOException {
+ long total = 0;
+ try (DataStreamInput in = getStreamInput(path, offset, length)) {
+ while (true) {
+ final ReferenceCountedObject<DataStreamReply> ref =
in.readAsync().join();
+ try {
+ final DataStreamReply reply = ref.get();
+ if (reply.getType() == Type.STREAM_HEADER) {
+ Preconditions.assertTrue(reply.isSuccess(),
+ () -> "Failed to stream read " + path + ", reply=" + reply);
+ return total;
+ } else {
+ Preconditions.assertTrue(reply.isSuccess(),
+ () -> "Failed to stream read " + path + ", reply=" +
reply);
+ Preconditions.assertEquals(Type.STREAM_DATA, reply.getType(),
+ "reply type for stream read " + path);
+ final ByteBuffer data = reply.nioBuffer();
+ while (data.hasRemaining()) {
+ total += channel.write(data);
+ }
+ }
+ } finally {
+ ref.release();
+ }
+ }
+ }
+ }
+
public CompletableFuture<Long> writeAsync(String path, long offset, boolean
close, ByteBuffer buffer, boolean sync) {
return writeImpl(this::sendAsync, path, offset, close, buffer, sync
).thenApply(reply -> JavaUtils.supplyAndWrapAsCompletionException(
diff --git
a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java
b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java
index d9a1463b9..4ff83369a 100644
---
a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java
+++
b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java
@@ -43,6 +43,7 @@ import
org.apache.ratis.thirdparty.com.google.protobuf.InvalidProtocolBufferExce
import org.apache.ratis.util.FileUtils;
import java.io.IOException;
+import java.nio.channels.WritableByteChannel;
import java.nio.file.Path;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicReference;
@@ -106,6 +107,24 @@ public class FileStoreStateMachine extends
BaseStateMachine {
.thenApply(reply -> Message.valueOf(reply.toByteString()));
}
+ @Override
+ public void query(Message request, WritableByteChannel stream) {
+ try {
+ final ReadRequestProto proto =
ReadRequestProto.parseFrom(request.getContent());
+ if (proto.getIsWatch()) {
+ throw new IOException("Watch is not supported for streaming read: " +
proto);
+ }
+ files.streamRead(proto.getPath().toStringUtf8(), proto.getOffset(),
proto.getLength(), stream);
+ } catch (Exception e) {
+ LOG.error(getId() + ": Failed streaming read for " + request, e);
+ try {
+ stream.close();
+ } catch (IOException ignored) {
+ // ignore
+ }
+ }
+ }
+
@Override
public TransactionContext startTransaction(RaftClientRequest request) throws
IOException {
final ByteString content = request.getMessage().getContent();
diff --git
a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/FileStore.java
b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/FileStore.java
index 9d50e3421..c62d5a10c 100644
---
a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/FileStore.java
+++
b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/FileStore.java
@@ -35,6 +35,7 @@ public final class FileStore {
commands.add(new Server());
commands.add(new LoadGen());
commands.add(new DataStream());
+ commands.add(new Read());
return commands;
}
}
diff --git
a/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/Read.java
b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/Read.java
new file mode 100644
index 000000000..128a9abba
--- /dev/null
+++
b/ratis-examples/src/main/java/org/apache/ratis/examples/filestore/cli/Read.java
@@ -0,0 +1,158 @@
+/*
+ * 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.examples.filestore.cli;
+
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.Parameters;
+import org.apache.ratis.RaftConfigKeys;
+import org.apache.ratis.client.RaftClient;
+import org.apache.ratis.client.RaftClientConfigKeys;
+import org.apache.ratis.conf.RaftProperties;
+import org.apache.ratis.datastream.SupportedDataStreamType;
+import org.apache.ratis.examples.common.SubCommandBase;
+import org.apache.ratis.examples.filestore.FileStoreClient;
+import org.apache.ratis.grpc.GrpcConfigKeys;
+import org.apache.ratis.grpc.GrpcFactory;
+import org.apache.ratis.protocol.ClientId;
+import org.apache.ratis.protocol.RaftGroup;
+import org.apache.ratis.protocol.RaftGroupId;
+import org.apache.ratis.rpc.SupportedRpcType;
+import org.apache.ratis.server.RaftServerConfigKeys;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.apache.ratis.util.FileUtils;
+import org.apache.ratis.util.SizeInBytes;
+import org.apache.ratis.util.TimeDuration;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Subcommand to read files from FileStore using streaming read.
+ */
+@Parameters(commandDescription = "Read files from FileStore using streaming
read")
+public class Read extends SubCommandBase {
+
+ @Parameter(names = {"--files"}, description = "Comma-separated file names on
the server", required = true)
+ private String files;
+
+ @Parameter(names = {"--size"}, description = "Number of bytes to read per
file", required = true)
+ private long fileSizeInBytes;
+
+ @Parameter(names = {"--offset"}, description = "Offset to start reading
from", required = false)
+ private long offset;
+
+ @Override
+ public void run() throws Exception {
+ final List<String> fileNames = Arrays.asList(files.split(","));
+ if (fileNames.isEmpty()) {
+ throw new IllegalArgumentException("No files specified in --files");
+ }
+
+ final RaftProperties raftProperties = newRaftProperties();
+
+ System.out.println("Starting streaming read now");
+
+ final long startTime = System.currentTimeMillis();
+ long totalReadBytes = 0;
+ final List<ReadResult> results = new ArrayList<>();
+ try (FileStoreClient client = newClient(raftProperties)) {
+ for (String fileName : fileNames) {
+ final String name = fileName.trim();
+ final ReadResult result = readFile(client, name);
+ results.add(result);
+ if (result.bytesRead != fileSizeInBytes) {
+ System.err.println("Error: file:" + name + " read:" +
result.bytesRead
+ + " mismatch expected size:" + fileSizeInBytes);
+ }
+ totalReadBytes += result.bytesRead;
+ }
+ }
+
+ final long endTime = System.currentTimeMillis();
+ System.out.println("Total files read: " + fileNames.size());
+ for (ReadResult result : results) {
+ System.out.println(" " + result.fileName + " -> " + result.tempFile
+ + " (" + result.bytesRead + " bytes)");
+ }
+ System.out.println("Each file size: " + fileSizeInBytes);
+ System.out.println("Total data read: " + totalReadBytes + " bytes");
+ System.out.println("Total time taken: " + (endTime - startTime) + "
millis");
+ }
+
+ private ReadResult readFile(FileStoreClient client, String fileName) throws
IOException {
+ final File tempFile = Files.createTempFile("filestore-read-", "-" +
fileName).toFile();
+ try (FileChannel out = FileUtils.newFileChannel(tempFile,
+ StandardOpenOption.CREATE, StandardOpenOption.WRITE,
StandardOpenOption.TRUNCATE_EXISTING)) {
+ final long bytesRead = client.streamRead(fileName, offset,
fileSizeInBytes, out);
+ return new ReadResult(fileName, tempFile.getAbsolutePath(), bytesRead);
+ } finally {
+ FileUtils.deleteFully(tempFile);
+ }
+ }
+
+ private static final class ReadResult {
+ private final String fileName;
+ private final String tempFile;
+ private final long bytesRead;
+
+ private ReadResult(String fileName, String tempFile, long bytesRead) {
+ this.fileName = fileName;
+ this.tempFile = tempFile;
+ this.bytesRead = bytesRead;
+ }
+ }
+
+ private static RaftProperties newRaftProperties() {
+ final int raftSegmentPreallocatedSize = 1024 * 1024 * 1024;
+ final RaftProperties raftProperties = new RaftProperties();
+ RaftConfigKeys.Rpc.setType(raftProperties, SupportedRpcType.GRPC);
+ GrpcConfigKeys.setMessageSizeMax(raftProperties,
SizeInBytes.valueOf(raftSegmentPreallocatedSize));
+ RaftServerConfigKeys.Log.Appender.setBufferByteLimit(raftProperties,
+ SizeInBytes.valueOf(raftSegmentPreallocatedSize));
+ RaftServerConfigKeys.Log.setWriteBufferSize(raftProperties,
+ SizeInBytes.valueOf(raftSegmentPreallocatedSize));
+ RaftServerConfigKeys.Log.setPreallocatedSize(raftProperties,
+ SizeInBytes.valueOf(raftSegmentPreallocatedSize));
+ RaftServerConfigKeys.Log.setSegmentSizeMax(raftProperties,
SizeInBytes.valueOf(raftSegmentPreallocatedSize));
+ RaftConfigKeys.DataStream.setType(raftProperties,
SupportedDataStreamType.NETTY);
+ RaftServerConfigKeys.Log.setSegmentCacheNumMax(raftProperties, 2);
+ RaftClientConfigKeys.Rpc.setRequestTimeout(raftProperties,
+ TimeDuration.valueOf(50000, TimeUnit.MILLISECONDS));
+ return raftProperties;
+ }
+
+ private FileStoreClient newClient(RaftProperties raftProperties) throws
IOException {
+ final RaftGroup raftGroup = RaftGroup.valueOf(
+ RaftGroupId.valueOf(ByteString.copyFromUtf8(getRaftGroupId())),
getPeers());
+ final RaftClient client = RaftClient.newBuilder()
+ .setProperties(raftProperties)
+ .setRaftGroup(raftGroup)
+ .setClientRpc(new GrpcFactory(new org.apache.ratis.conf.Parameters())
+ .newRaftClientRpc(ClientId.randomId(), raftProperties))
+ .setPrimaryDataStreamServer(getPeers()[0])
+ .build();
+ return new FileStoreClient(client);
+ }
+}
diff --git
a/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreStreamingBaseTest.java
b/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreStreamingBaseTest.java
index cdcee0ef0..d70efaaa9 100644
---
a/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreStreamingBaseTest.java
+++
b/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreStreamingBaseTest.java
@@ -104,6 +104,42 @@ public abstract class FileStoreStreamingBaseTest <CLUSTER
extends MiniRaftCluste
cluster.shutdown();
}
+ @Test
+ public void testFileStoreStreamReadAfterStreamWrite() throws Exception {
+ final CLUSTER cluster = newCluster(NUM_PEERS);
+ cluster.start();
+ RaftTestUtil.waitForLeader(cluster);
+
+ final RaftGroup raftGroup = cluster.getGroup();
+ final Collection<RaftPeer> peers = raftGroup.getPeers();
+ final RaftPeer primary = cluster.getLeader().getPeer();
+
+ final CheckedSupplier<FileStoreClient, IOException> newClient =
+ () -> new FileStoreClient(cluster.getGroup(), getProperties(),
primary);
+
+ final RoutingTable routingTable =
DataStreamTestUtils.getRoutingTableChainTopology(peers, primary);
+ testSingleFileReadAfterStreamWrite("foo", SizeInBytes.valueOf("2M"),
10_000, newClient, routingTable);
+
+ cluster.shutdown();
+ }
+
+ private static void testSingleFileReadAfterStreamWrite(
+ String path, SizeInBytes fileLength, int bufferSize,
+ CheckedSupplier<FileStoreClient, IOException> newClient, RoutingTable
routingTable)
+ throws Exception {
+ LOG.info("testSingleFileAfterStreamWrite with path={}, fileLength={}",
path, fileLength);
+ FileStoreWriter.newBuilder()
+ .setFileName(path)
+ .setFileSize(fileLength)
+ .setBufferSize(bufferSize)
+ .setFileStoreClientSupplier(newClient)
+ .build()
+ .streamWrite(routingTable)
+ .streamRead()
+ .delete()
+ .close();
+ }
+
private void testSingleFile(
String path, SizeInBytes fileLength, int bufferSize,
CheckedSupplier<FileStoreClient, IOException> newClient,
RoutingTable routingTable)
diff --git
a/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreWriter.java
b/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreWriter.java
index 480fe40d1..18f8f89ab 100644
---
a/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreWriter.java
+++
b/ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreWriter.java
@@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.channels.WritableByteChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
@@ -133,7 +134,7 @@ final class FileStoreWriter implements Closeable {
return this;
}
- public FileStoreWriter streamWriteAndVerify(RoutingTable routingTable) {
+ public FileStoreWriter streamWrite(RoutingTable routingTable) {
final int size = fileSize.getSizeInt();
final DataStreamOutput dataStreamOutput = client.getStreamOutput(fileName,
size, routingTable);
final List<CompletableFuture<DataStreamReply>> futures = new ArrayList<>();
@@ -156,8 +157,6 @@ final class FileStoreWriter implements Closeable {
DataStreamReply reply = dataStreamOutput.closeAsync().join();
Assertions.assertTrue(reply.isSuccess());
- // TODO: handle when any of the writeAsync has failed.
- // check writeAsync requests
for (int i = 0; i < futures.size(); i++) {
reply = futures.get(i).join();
Assertions.assertTrue(reply.isSuccess());
@@ -168,6 +167,37 @@ final class FileStoreWriter implements Closeable {
return this;
}
+ public FileStoreWriter streamWriteAndVerify(RoutingTable routingTable) {
+ return streamWrite(routingTable);
+ }
+
+ FileStoreWriter streamRead() throws IOException {
+ final long expected = fileSize.getSizeInt();
+ final long bytesRead = client.streamRead(fileName, 0, expected, DISCARD);
+ Assertions.assertEquals(expected, bytesRead,
+ () -> "stream read " + fileName + ": expected " + expected + " bytes");
+ LOG.info("Stream read successful: {} bytes from {}", bytesRead, fileName);
+ return this;
+ }
+
+ private static final WritableByteChannel DISCARD = new WritableByteChannel()
{
+ @Override
+ public int write(ByteBuffer src) {
+ final int n = src.remaining();
+ src.position(src.limit());
+ return n;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return true;
+ }
+
+ @Override
+ public void close() {
+ }
+ };
+
CompletableFuture<FileStoreWriter> writeAsync(boolean sync) {
Objects.requireNonNull(asyncExecutor, "asyncExecutor == null");
final Random r = new Random(seed);
