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/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 07e64df5c6f HDDS-15424. Support concurrent positional read (#11102)
07e64df5c6f is described below
commit 07e64df5c6f3c2196848c73f2598e2056ff5ebf5
Author: Tak Lon (Stephen) Wu <[email protected]>
AuthorDate: Tue Sep 15 11:01:43 2026 -0700
HDDS-15424. Support concurrent positional read (#11102)
---
.../hadoop/hdds/scm/storage/BlockInputStream.java | 120 +++++++++++++++-
.../hadoop/hdds/scm/storage/ChunkInputStream.java | 151 +++++++++++++++++----
.../hdds/scm/storage/LocalChunkInputStream.java | 48 +++++--
.../hdds/scm/storage/MultipartInputStream.java | 73 +++++++++-
.../hdds/scm/storage/DummyChunkInputStream.java | 22 +--
.../hdds/scm/storage/PositionedReadTestHelper.java | 92 +++++++++++++
.../hdds/scm/storage/TestBlockInputStream.java | 109 +++++++++++++++
.../hdds/scm/storage/TestChunkInputStream.java | 71 ++++++++++
.../hdds/scm/storage/TestMultipartInputStream.java | 126 +++++++++++++++++
9 files changed, 752 insertions(+), 60 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java
index 6d5ac9c8cf2..57757cb79f7 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java
@@ -23,6 +23,7 @@
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -77,7 +78,7 @@ public class BlockInputStream extends
BlockExtendedInputStream {
private XceiverClientShortCircuit xceiverClientShortCircuit;
private final AtomicBoolean fallbackToGrpc = new AtomicBoolean(false);
private volatile FileInputStream blockFileInputStream;
- private boolean initialized = false;
+ private volatile boolean initialized = false;
// TODO: do we need to change retrypolicy based on exception.
private final RetryPolicy retryPolicy;
@@ -412,10 +413,10 @@ protected synchronized int
readWithStrategy(ByteReaderStrategy strategy)
// Get the current chunkStream and read data from it
ChunkInputStream current = chunkStreams.get(chunkIndex);
- int numBytesToRead = Math.min(len, (int)current.getRemaining());
+ int readLength = Math.min(len, (int)current.getRemaining());
int numBytesRead;
try {
- numBytesRead = strategy.readFromBlock(current, numBytesToRead);
+ numBytesRead = strategy.readFromBlock(current, readLength);
retries = 0;
// If we get a StorageContainerException or an IOException due to
// datanodes are not reachable, refresh to get the latest pipeline
@@ -443,14 +444,14 @@ protected synchronized int
readWithStrategy(ByteReaderStrategy strategy)
}
}
- if (numBytesRead != numBytesToRead) {
+ if (numBytesRead != readLength) {
// This implies that there is either data loss or corruption in the
// chunk entries. Even EOF in the current stream would be covered in
// this case.
throw new IOException(String.format(
- "Inconsistent read for chunkName=%s length=%d numBytesToRead= %d "
+
+ "Inconsistent read for chunkName=%s length=%d readLength= %d " +
"numBytesRead=%d", current.getChunkName(), current.getLength(),
- numBytesToRead, numBytesRead));
+ readLength, numBytesRead));
}
totalReadLen += numBytesRead;
len -= numBytesRead;
@@ -462,6 +463,111 @@ protected synchronized int
readWithStrategy(ByteReaderStrategy strategy)
return totalReadLen;
}
+ /**
+ * Stateless positioned read across this block's chunks. Fills up to
+ * {@code dst.remaining()} bytes starting from {@code blockRelativePosition}
+ * without mutating this stream's cursor ({@code chunkIndex},
+ * {@code blockPosition}) or the sequential chunk streams' buffered state.
+ * Each covering chunk is read through an ephemeral {@link ChunkInputStream}
+ * closed as soon as its bytes have been copied.
+ *
+ * @return bytes copied into {@code dst}, or {@link #EOF} at EOF
+ */
+ int readPositioned(long blockRelativePosition, ByteBuffer dst) throws
IOException {
+ if (blockRelativePosition < 0) {
+ return EOF;
+ }
+
+ if (!initialized) {
+ initialize();
+ }
+ final long[] offsets;
+ final BlockData currentBlockData;
+ final long blockLength;
+ synchronized (this) {
+ checkOpen();
+ if (chunkOffsets == null || blockData == null || blockRelativePosition
>= length) {
+ return EOF;
+ }
+
+ offsets = Arrays.copyOf(chunkOffsets, chunkOffsets.length);
+ currentBlockData = blockData;
+ blockLength = length;
+ }
+
+ final List<ChunkInfo> chunkInfos = currentBlockData.getChunksList();
+ int index = MultipartInputStream.binarySearchOffsetIndex(offsets,
blockRelativePosition);
+
+ long pos = blockRelativePosition;
+ int totalReadLen = 0;
+ while (dst.hasRemaining() && pos < blockLength && index <
chunkInfos.size()) {
+ final ChunkInfo chunkInfo = chunkInfos.get(index);
+ final long chunkOffset = pos - offsets[index];
+ final long readLength = Math.min(Math.min(dst.remaining(),
chunkInfo.getLen() - chunkOffset), blockLength - pos);
+ if (readLength > 0) {
+ final int numBytesRead = readChunkAt(chunkInfo, chunkOffset, (int)
readLength, dst);
+ totalReadLen += numBytesRead;
+ pos += numBytesRead;
+ }
+ index++;
+ }
+ return totalReadLen == 0 ? EOF : totalReadLen;
+ }
+
+ /**
+ * Read {@code readLength} bytes starting at {@code chunkOffset} of the
given chunk into {@code dst}
+ * through an ephemeral {@link ChunkInputStream}, retrying like {@link
#readWithStrategy(ByteReaderStrategy)}
+ * but with a retry counter local to this call.
+ */
+ private int readChunkAt(ChunkInfo chunkInfo, long chunkOffset, int
readLength, ByteBuffer dst) throws IOException {
+ final int startPosition = dst.position();
+ int preadRetries = 0;
+ while (true) {
+ final ChunkInputStream chunkStream;
+ synchronized (this) {
+ checkOpen();
+ chunkStream = createChunkInputStream(chunkInfo);
+ }
+ final int numBytesRead;
+ try {
+ final int oldLimit = dst.limit();
+ try {
+ dst.limit(startPosition + readLength);
+ numBytesRead = chunkStream.readPositioned(chunkOffset, dst);
+ } finally {
+ dst.limit(oldLimit);
+ }
+ } catch (SCMSecurityException ex) {
+ throw ex;
+ } catch (StorageContainerException ex) {
+ if (!shouldRetryRead(ex, retryPolicy, ++preadRetries)) {
+ throw ex;
+ }
+ handleReadError(ex);
+ dst.position(startPosition);
+ continue;
+ } catch (IOException ex) {
+ if (!shouldRetryRead(ex, retryPolicy, ++preadRetries)) {
+ throw ex;
+ }
+ if (isConnectivityIssue(ex)) {
+ handleReadError(ex);
+ }
+ dst.position(startPosition);
+ continue;
+ } finally {
+ chunkStream.close();
+ }
+
+ if (numBytesRead != readLength) {
+ throw new IOException(String.format(
+ "Inconsistent read for chunkName=%s length=%d readLength=%d
numBytesRead=%d",
+ chunkInfo.getChunkName(), chunkInfo.getLen(), readLength,
numBytesRead));
+ }
+ return numBytesRead;
+ }
+ }
+
/**
* Seeks the BlockInputStream to the specified position. If the stream is
* not initialized, save the seeked position via blockPosition. Otherwise,
@@ -630,7 +736,7 @@ private synchronized void storePosition() {
blockPosition = getPos();
}
- private void handleReadError(IOException cause) throws IOException {
+ private synchronized void handleReadError(IOException cause) throws
IOException {
releaseClient();
final List<ChunkInputStream> inputStreams = this.chunkStreams;
if (inputStreams != null) {
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
index ee70a67dd85..f78adc85e41 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
@@ -26,6 +26,7 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.Objects;
import java.util.function.Supplier;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.fs.ByteBufferReadable;
@@ -37,6 +38,7 @@
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
+import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.DatanodeBlockID;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadChunkResponseProto;
import org.apache.hadoop.hdds.scm.XceiverClientFactory;
import org.apache.hadoop.hdds.scm.XceiverClientSpi;
@@ -47,6 +49,7 @@
import org.apache.hadoop.ozone.common.OzoneChecksumException;
import org.apache.hadoop.ozone.common.utils.BufferUtils;
import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
/**
@@ -300,10 +303,17 @@ private void updateDatanodeBlockId(Pipeline pipeline)
throws IOException {
datanodeBlockID =
blockID.getDatanodeBlockIDProtobufBuilder(replicaIdx).build();
}
+ private DatanodeBlockID buildDatanodeBlockId(Pipeline pipeline) throws
IOException {
+ updateDatanodeBlockId(pipeline);
+ return datanodeBlockID;
+ }
+
/**
* Acquire new client if previous one was released.
+ *
+ * @return the held client after ensuring it is acquired
*/
- protected synchronized void acquireClient() throws IOException {
+ private synchronized void acquireClient() throws IOException {
if (xceiverClientFactory != null && xceiverClient == null) {
Pipeline pipeline = pipelineSupplier.get();
xceiverClient = xceiverClientFactory.acquireClientForReadData(pipeline);
@@ -374,29 +384,10 @@ private synchronized void readChunkFromContainer(int len)
throws IOException {
// successful read in adjustBufferPosition()
storePosition();
- long adjustedBuffersOffset, adjustedBuffersLen;
- if (verifyChecksum) {
- // Adjust the chunk offset and length to include required checksum
- // boundaries
- Pair<Long, Long> adjustedOffsetAndLength =
- computeChecksumBoundaries(startByteIndex, len);
- adjustedBuffersOffset = adjustedOffsetAndLength.getLeft();
- adjustedBuffersLen = adjustedOffsetAndLength.getRight();
- } else {
- // Read from the startByteIndex
- adjustedBuffersOffset = startByteIndex;
- adjustedBuffersLen = len;
- }
-
- // Adjust the chunkInfo so that only the required bytes are read from
- // the chunk.
- final ChunkInfo adjustedChunkInfo = ChunkInfo.newBuilder(chunkInfo)
- .setOffset(chunkInfo.getOffset() + adjustedBuffersOffset)
- .setLen(adjustedBuffersLen)
- .build();
+ final ChunkInfo adjustedChunkInfo = getChunkInfo(startByteIndex, len);
readChunkDataIntoBuffers(adjustedChunkInfo);
- bufferOffsetWrtChunkData = adjustedBuffersOffset;
+ bufferOffsetWrtChunkData = adjustedChunkInfo.getOffset() -
chunkInfo.getOffset();
// If the stream was seeked to position before, then the buffer
// position should be adjusted as the reads happen at checksum boundaries.
@@ -408,6 +399,24 @@ private synchronized void readChunkFromContainer(int len)
throws IOException {
adjustBufferPosition(startByteIndex - bufferOffsetWrtChunkData);
}
+ ChunkInfo getChunkInfo(long chunkRelativePosition, int toRead) {
+ final long adjustedOffset;
+ final long adjustedLen;
+ if (verifyChecksum) {
+ Pair<Long, Long> boundaries =
computeChecksumBoundaries(chunkRelativePosition, toRead);
+ adjustedOffset = boundaries.getLeft();
+ adjustedLen = boundaries.getRight();
+ } else {
+ adjustedOffset = chunkRelativePosition;
+ adjustedLen = toRead;
+ }
+
+ return ChunkInfo.newBuilder(chunkInfo)
+ .setOffset(chunkInfo.getOffset() + adjustedOffset)
+ .setLen(adjustedLen)
+ .build();
+ }
+
protected void readChunkDataIntoBuffers(ChunkInfo readChunkInfo)
throws IOException {
buffers = readChunk(readChunkInfo);
@@ -426,15 +435,101 @@ protected void readChunkDataIntoBuffers(ChunkInfo
readChunkInfo)
}
/**
- * Send RPC call to get the chunk from the container.
+ * Stateless positioned read of up to {@code dst.remaining()} bytes starting
+ * at {@code chunkRelativePosition} within this chunk. Unlike the buffered
+ * {@link #read} path, this does not read or mutate any of the instance's
+ * buffer/position state ({@code buffers}, {@code chunkPosition},
+ * {@code bufferOffsetWrtChunkData}, ...), so it is safe to call concurrently
+ * from multiple threads sharing the same stream.
+ *
+ * @param chunkRelativePosition start offset within this chunk
+ * @param dst destination buffer
+ * @return number of bytes copied into {@code dst}, or {@link #EOF} at EOF
+ */
+ int readPositioned(long chunkRelativePosition, ByteBuffer dst) throws
IOException {
+ if (chunkRelativePosition < 0 || chunkRelativePosition >= length) {
+ return EOF;
+ }
+ final int toRead = (int) Math.min(dst.remaining(), length -
chunkRelativePosition);
+ if (toRead == 0) {
+ return 0;
+ }
+
+ final ChunkInfo readChunkInfo = getChunkInfo(chunkRelativePosition,
toRead);
+ final long adjustedOffset = readChunkInfo.getOffset() -
chunkInfo.getOffset();
+
+ final long skip = chunkRelativePosition - adjustedOffset;
+ if (xceiverClientFactory == null) {
+ // for LocalChunkInputStream, which does not use a xceiver client, just
read the chunk directly
+ return copyRange(readChunk(readChunkInfo), skip, toRead, dst);
+ }
+
+ final Pipeline pipeline = pipelineSupplier.get();
+ final ContainerProtos.DatanodeBlockID bid = buildDatanodeBlockId(pipeline);
+ final XceiverClientSpi client =
xceiverClientFactory.acquireClientForReadData(pipeline);
+ try {
+ final ByteBuffer[] readBuffers = readChunk(readChunkInfo, client, bid,
validators, tokenSupplier.get());
+ return copyRange(readBuffers, skip, toRead, dst);
+ } finally {
+ xceiverClientFactory.releaseClientForReadData(client, false);
+ }
+ }
+
+ /**
+ * Copy {@code toCopy} bytes from {@code src} buffers, skipping the first
+ * {@code skip} bytes, into {@code dst}. Operates on duplicates so the source
+ * buffers' positions are left untouched.
+ */
+ private static int copyRange(ByteBuffer[] src, long skip, int toCopy,
ByteBuffer dst) {
+ long remainingSkip = skip;
+ int copied = 0;
+ for (ByteBuffer buffer : src) {
+ if (copied >= toCopy) {
+ break;
+ }
+ if (remainingSkip > 0) {
+ if (remainingSkip >= buffer.remaining()) {
+ remainingSkip -= buffer.remaining();
+ continue;
+ } else {
+ buffer.position(Math.toIntExact(buffer.position() + remainingSkip));
+ remainingSkip = 0;
+ }
+ }
+ int n = Math.min(buffer.remaining(), toCopy - copied);
+ if (n <= 0) {
+ continue;
+ }
+ buffer.limit(buffer.position() + n);
+ dst.put(buffer);
+ copied += n;
+ }
+ return copied;
+ }
+
+ /**
+ * Send RPC call to get the chunk from the container using the
sequential-read client held in
+ * {@link #xceiverClient}. Called by the buffered sequential read path via
+ * {@link #readChunkDataIntoBuffers}.
*/
@VisibleForTesting
- protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo)
- throws IOException {
+ protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo) throws IOException
{
+ return readChunk(readChunkInfo, xceiverClient, datanodeBlockID,
validators, tokenSupplier.get());
+ }
- ReadChunkResponseProto readChunkResponse =
- ContainerProtocolCalls.readChunk(xceiverClient, readChunkInfo,
datanodeBlockID, validators,
- tokenSupplier.get());
+ /**
+ * Send RPC call to get the chunk from the container using an explicitly
provided client.
+ * Used by the positioned-read path so callers hold a local reference to the
client
+ * rather than re-reading the shared {@link #xceiverClient} field after the
lock is released.
+ * Subclasses that do not use an xceiver client (e.g. local short-circuit
reads) should
+ * override this method to ignore {@code client} and read through their own
mechanism.
+ */
+ private static ByteBuffer[] readChunk(
+ ChunkInfo chunk, XceiverClientSpi client, DatanodeBlockID blockID,
+ List<Validator> validators, Token<? extends TokenIdentifier> token)
throws IOException {
+ Objects.requireNonNull(client, "client");
+ final ReadChunkResponseProto readChunkResponse =
ContainerProtocolCalls.readChunk(
+ client, chunk, blockID, validators, token);
if (readChunkResponse.hasData()) {
return readChunkResponse.getData().asReadOnlyByteBufferList()
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java
index 9de58ac7fe9..a6e429fd6e3 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/LocalChunkInputStream.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdds.scm.storage;
import com.google.common.annotations.VisibleForTesting;
+import java.io.EOFException;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
@@ -62,7 +63,9 @@ public class LocalChunkInputStream extends ChunkInputStream
LocalChunkInputStream(ChunkInfo chunkInfo, BlockID blockId,
XceiverClientFactory xceiverClientFactory,
Supplier<Pipeline> pipelineSupplier, boolean verifyChecksum,
Supplier<Token<?>> tokenSupplier,
XceiverClientShortCircuit xceiverClientShortCircuit, FileInputStream
blockInputStream) {
- super(chunkInfo, blockId, xceiverClientFactory, pipelineSupplier,
verifyChecksum, tokenSupplier);
+ // Pass null for xceiverClientFactory such readPositioned() is taken, both
routing reads
+ // and checksum verification are done locally.
+ super(chunkInfo, blockId, null, pipelineSupplier, verifyChecksum,
tokenSupplier);
this.chunkInfo = chunkInfo;
this.dataIn = blockInputStream.getChannel();
this.validator = this::validateChunk;
@@ -80,14 +83,43 @@ public class LocalChunkInputStream extends ChunkInputStream
protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo)
throws IOException {
int bytesPerChecksum = chunkInfo.getChecksumData().getBytesPerChecksum();
- final ByteBuffer[] buffers =
BufferUtils.assignByteBuffers(readChunkInfo.getLen(),
- bytesPerChecksum);
- dataIn.position(readChunkInfo.getOffset()).read(buffers);
+ final ByteBuffer[] buffers =
BufferUtils.assignByteBuffers(readChunkInfo.getLen(), bytesPerChecksum);
+ readAtOffset(buffers, readChunkInfo.getOffset());
Arrays.stream(buffers).forEach(ByteBuffer::flip);
validator.accept(Arrays.asList(buffers), readChunkInfo);
return buffers;
}
+ /**
+ * Read into {@code buffers} starting at {@code fileOffset} using positional
+ * {@link FileChannel} reads so concurrent callers on different chunks (which
+ * share the same underlying channel) do not stomp each other's cursor.
+ */
+ private void readAtOffset(ByteBuffer[] buffers, long fileOffset) throws
IOException {
+ long pos = fileOffset;
+ for (ByteBuffer buffer : buffers) {
+ final int remaining = buffer.remaining();
+ final long read = readAtOffset(buffer, pos);
+ if (buffer.hasRemaining()) {
+ throw new EOFException("Read only " + read + "bytes but expected to
read " + remaining
+ + " bytes at offset " + pos + " for chunk " +
chunkInfo.getChunkName());
+ }
+ pos += read;
+ }
+ }
+
+ private long readAtOffset(ByteBuffer buffer, long fileOffset) throws
IOException {
+ long pos = fileOffset;
+ while (buffer.hasRemaining()) {
+ final int n = dataIn.read(buffer, pos);
+ if (n < 0) {
+ break;
+ }
+ pos += n;
+ }
+ return pos - fileOffset;
+ }
+
private void validateChunk(List<ByteBuffer> bufferList, ChunkInfo
readChunkInfo)
throws OzoneChecksumException {
if (verifyChecksum) {
@@ -105,12 +137,4 @@ private void validateChunk(List<ByteBuffer> bufferList,
ChunkInfo readChunkInfo)
Checksum.verifyChecksum(bufferList, startIndex, checksumData);
}
}
-
- /**
- * Acquire short-circuit local read client.
- */
- @Override
- protected synchronized void acquireClient() throws IOException {
- // do nothing, read data doesn't need short-circuit client
- }
}
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java
index e7c259ce464..584d97c778e 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java
@@ -40,6 +40,7 @@ public class MultipartInputStream extends ExtendedInputStream
{
// List of PartInputStream, one for each part of the key
private final List<? extends PartInputStream> partStreams;
private final boolean isStreamBlockInputStream;
+ private final boolean statelessPositionedReadSupported;
// partOffsets[i] stores the index of the first data byte in
// partStream w.r.t the whole key data.
@@ -70,15 +71,19 @@ public MultipartInputStream(String keyName,
// Calculate and update the partOffsets
this.partOffsets = new long[inputStreams.size()];
+ boolean statelessSupported = !inputStreams.isEmpty();
int i = 0;
long streamLength = 0L;
for (PartInputStream partInputStream : inputStreams) {
this.partOffsets[i++] = streamLength;
if (isStreamBlockInputStream) {
Preconditions.assertInstanceOf(partInputStream,
StreamBlockInputStream.class);
+ } else if (statelessSupported && !(partInputStream instanceof
BlockInputStream)) {
+ statelessSupported = false;
}
streamLength += partInputStream.getLength();
}
+ this.statelessPositionedReadSupported = statelessSupported;
this.length = streamLength;
}
@@ -191,10 +196,19 @@ public synchronized void seek(long pos) throws
IOException {
@Override
public boolean readFully(long position, ByteBuffer buffer) throws
IOException {
- if (!isStreamBlockInputStream) {
- return false;
+ if (isStreamBlockInputStream) {
+ return readFullyStreamBlock(position, buffer);
}
+ return readFullyStateless(position, buffer);
+ }
+ /**
+ * Positioned read for the StreamBlock path. This emulates a positioned read
+ * with seek-read-restore on the shared stream cursor, so it is synchronized
+ * to keep concurrent callers from corrupting each other's position.
+ */
+ private synchronized boolean readFullyStreamBlock(long position,
+ ByteBuffer buffer) throws IOException {
final long oldPos = getPos();
seek(position);
try {
@@ -219,6 +233,61 @@ int readImpl(InputStream inputStream) throws IOException {
return true;
}
+ /**
+ * Stateless positioned read for the replicated (Ratis) path. Routes the read
+ * across the part {@link BlockInputStream}s using the immutable
+ * {@link #partOffsets} without seeking or mutating the shared cursor, so
+ * concurrent positioned reads run independently. Returns {@code false} (so
+ * the caller can fall back) when any part is not a {@link BlockInputStream},
+ * e.g. erasure coded parts.
+ */
+ private boolean readFullyStateless(long position, ByteBuffer buffer) throws
IOException {
+ if (!buffer.hasRemaining()) {
+ return true;
+ }
+ if (!statelessPositionedReadSupported) {
+ return false;
+ }
+
+ long pos = position;
+ int bytesRead = 0;
+ while (buffer.hasRemaining()) {
+ if (pos < 0 || pos >= length) {
+ if (bytesRead > 0) {
+ return true;
+ }
+ throw new EOFException("EOF encountered at pos: " + pos +
+ " for key: " + key);
+ }
+ int idx = binarySearchOffsetIndex(partOffsets, pos);
+ BlockInputStream part = (BlockInputStream) partStreams.get(idx);
+ long partPos = pos - partOffsets[idx];
+ int n = part.readPositioned(partPos, buffer);
+ if (n <= 0) {
+ if (bytesRead > 0) {
+ return true;
+ }
+ throw new EOFException("EOF encountered at pos: " + pos +
+ " for key: " + key);
+ }
+ bytesRead += n;
+ pos += n;
+ }
+ return true;
+ }
+
+ static int binarySearchOffsetIndex(long[] offsets, long pos) {
+ final int idx = Arrays.binarySearch(offsets, pos);
+ if (idx >= 0) {
+ // exact match: pos falls exactly on the start of chunk/part idx
+ return idx;
+ }
+ // binarySearch returns n = -insertionPoint - 1;
+ // insertionPoint is -n - 1
+ // the containing index is insertionPoint - 1.
+ return -idx - 2;
+ }
+
public synchronized void initialize() throws IOException {
// Pre-check that the stream has not been intialized already
if (initialized) {
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java
index 1ef6327c2a1..97ec08f4d12 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java
@@ -19,7 +19,9 @@
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.hadoop.hdds.client.BlockID;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
import org.apache.hadoop.hdds.scm.XceiverClientFactory;
@@ -34,8 +36,8 @@ public class DummyChunkInputStream extends ChunkInputStream {
private final byte[] chunkData;
- // Stores the read chunk data in each readChunk call
- private final List<ByteString> readByteBuffers = new ArrayList<>();
+ // Buffers from the most recent readChunk call.
+ private final AtomicReference<List<ByteString>> lastReadByteBuffers = new
AtomicReference<>();
public DummyChunkInputStream(ChunkInfo chunkInfo,
BlockID blockId,
@@ -54,7 +56,7 @@ protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo) {
int bufferCapacity = readChunkInfo.getChecksumData().getBytesPerChecksum();
int bufferLen;
- readByteBuffers.clear();
+ List<ByteString> chunkBuffers = new ArrayList<>();
while (remainingToRead > 0) {
if (remainingToRead < bufferCapacity) {
bufferLen = remainingToRead;
@@ -64,19 +66,16 @@ protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo) {
ByteString byteString = ByteString.copyFrom(chunkData,
offset, bufferLen);
- readByteBuffers.add(byteString);
+ chunkBuffers.add(byteString);
offset += bufferLen;
remainingToRead -= bufferLen;
}
- return BufferUtils.getReadOnlyByteBuffers(readByteBuffers)
- .toArray(new ByteBuffer[0]);
- }
+ lastReadByteBuffers.set(chunkBuffers);
- @Override
- protected void acquireClient() {
- // No action needed
+ return BufferUtils.getReadOnlyByteBuffers(chunkBuffers)
+ .toArray(new ByteBuffer[0]);
}
@Override
@@ -85,6 +84,7 @@ protected void releaseClient() {
}
public List<ByteString> getReadByteBuffers() {
- return readByteBuffers;
+ List<ByteString> last = lastReadByteBuffers.get();
+ return last != null ? last : Collections.emptyList();
}
}
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/PositionedReadTestHelper.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/PositionedReadTestHelper.java
new file mode 100644
index 00000000000..fe5514698ff
--- /dev/null
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/PositionedReadTestHelper.java
@@ -0,0 +1,92 @@
+/*
+ * 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.hadoop.hdds.scm.storage;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Shared helpers for concurrent positioned-read unit tests.
+ */
+public final class PositionedReadTestHelper {
+
+ public static final int THREAD_COUNT = 8;
+ public static final int ITERATIONS = 100;
+ public static final int BUFFER_SIZE = 4096;
+ public static final int SOURCE_SIZE = 512 * 1024;
+
+ /**
+ * Performs a positioned read at the given key-relative offset into {@code
buf}.
+ */
+ @FunctionalInterface
+ public interface PositionedReadAction {
+ void readAtOffset(int offset, ByteBuffer buf) throws Exception;
+ }
+
+ private PositionedReadTestHelper() {
+ }
+
+ public static void runConcurrentPositionedReads(byte[] source,
+ PositionedReadAction action) throws Exception {
+ ExecutorService pool = Executors.newFixedThreadPool(THREAD_COUNT);
+ try {
+ List<Future<?>> futures = new ArrayList<>();
+ for (int t = 0; t < THREAD_COUNT; t++) {
+ final int threadId = t;
+ futures.add(pool.submit((Callable<Void>) () -> {
+ for (int i = 0; i < ITERATIONS; i++) {
+ int offset = (threadId * 1000 + i * 17) % (source.length -
BUFFER_SIZE);
+ ByteBuffer buf = ByteBuffer.allocate(BUFFER_SIZE);
+ action.readAtOffset(offset, buf);
+ buf.flip();
+ byte[] expected = Arrays.copyOfRange(source, offset, offset +
BUFFER_SIZE);
+ byte[] actual = new byte[BUFFER_SIZE];
+ buf.get(actual);
+ assertArrayEquals(expected, actual,
+ "thread " + threadId + " offset " + offset);
+ }
+ return null;
+ }));
+ }
+ for (Future<?> future : futures) {
+ future.get(1, TimeUnit.MINUTES);
+ }
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+
+ public static Throwable unwrapExecutionException(
+ ExecutionException executionException) {
+ Throwable cause = executionException.getCause();
+ while (cause instanceof RuntimeException && cause.getCause() != null) {
+ cause = cause.getCause();
+ }
+ return cause;
+ }
+}
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java
index 1b5e7667e84..baf4d7fef09 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java
@@ -18,6 +18,8 @@
package org.apache.hadoop.hdds.scm.storage;
import static
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CONTAINER_NOT_FOUND;
+import static
org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.BUFFER_SIZE;
+import static
org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.SOURCE_SIZE;
import static
org.apache.hadoop.hdds.scm.storage.TestChunkInputStream.generateRandomData;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -26,6 +28,7 @@
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.anyLong;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.reset;
@@ -312,6 +315,71 @@ void refreshesPipelineOnReadFailure(IOException ex) throws
Exception {
id -> newBlockLocationInfo);
}
+ @ParameterizedTest
+ @MethodSource("exceptionsTriggersRefresh")
+ void refreshesPipelineOnPositionedReadFailure(IOException ex) throws
Exception {
+ Pipeline pipeline = MockPipeline.createSingleNodePipeline();
+ BlockLocationInfo blockLocationInfo = mock(BlockLocationInfo.class);
+ when(blockLocationInfo.getPipeline()).thenReturn(pipeline);
+ Pipeline newPipeline = MockPipeline.createSingleNodePipeline();
+ BlockLocationInfo newBlockLocationInfo = mock(BlockLocationInfo.class);
+
+ testRefreshesPipelineOnPositionedReadFailure(ex, blockLocationInfo,
+ id -> newBlockLocationInfo);
+
+ when(newBlockLocationInfo.getPipeline()).thenReturn(newPipeline);
+ testRefreshesPipelineOnPositionedReadFailure(ex, blockLocationInfo,
+ id -> blockLocationInfo);
+
+ when(newBlockLocationInfo.getPipeline()).thenReturn(null);
+ testRefreshesPipelineOnPositionedReadFailure(ex, blockLocationInfo,
+ id -> newBlockLocationInfo);
+ }
+
+ private void testRefreshesPipelineOnPositionedReadFailure(IOException ex,
+ BlockLocationInfo blockLocationInfo,
+ Function<BlockID, BlockLocationInfo> refreshPipelineFunction)
+ throws Exception {
+ BlockID blockID = new BlockID(new ContainerBlockID(1, 1));
+ final int len = 200;
+ final ChunkInputStream stream =
throwingChunkInputStreamForPositionedRead(ex, len, true);
+
+ when(this.refreshFunction.apply(any()))
+ .thenAnswer(inv -> refreshPipelineFunction.apply(blockID));
+
+ try (BlockInputStream subject = createSubject(blockID,
+ blockLocationInfo.getPipeline(), stream)) {
+ subject.initialize();
+ ByteBuffer buf = ByteBuffer.allocate(len);
+ int bytesRead = subject.readPositioned(0, buf);
+ assertEquals(len, bytesRead);
+ verify(this.refreshFunction).apply(blockID);
+ } finally {
+ reset(this.refreshFunction);
+ }
+ }
+
+ private static ChunkInputStream throwingChunkInputStreamForPositionedRead(
+ IOException ex, int len, boolean succeedOnRetry) throws IOException {
+ final ChunkInputStream stream = mock(ChunkInputStream.class);
+ OngoingStubbing<Integer> stubbing =
+ when(stream.readPositioned(anyLong(), any(ByteBuffer.class)))
+ .thenThrow(ex);
+ if (succeedOnRetry) {
+ stubbing.thenAnswer(invocation -> {
+ ByteBuffer buffer = invocation.getArgument(1);
+ int n = Math.min(len, buffer.remaining());
+ for (int i = 0; i < n; i++) {
+ buffer.put((byte) 0);
+ }
+ return n;
+ });
+ }
+ when(stream.getRemaining()).thenReturn((long) len);
+ when(stream.getLength()).thenReturn((long) len);
+ return stream;
+ }
+
private void testRefreshesPipelineOnReadFailure(IOException ex,
BlockLocationInfo blockLocationInfo,
Function<BlockID, BlockLocationInfo> refreshPipelineFunction)
@@ -469,4 +537,45 @@ private static Stream<Arguments>
exceptionsTriggersRefresh() {
new StatusException(Status.UNAVAILABLE))))
);
}
+
+ @Test
+ public void testConcurrentPositionedRead() throws Exception {
+ try (BlockInputStream largeBlockStream = createLargeBlockStream()) {
+ largeBlockStream.initialize();
+ PositionedReadTestHelper.runConcurrentPositionedReads(blockData,
+ (offset, buf) -> {
+ int read = largeBlockStream.readPositioned(offset, buf);
+ if (read != BUFFER_SIZE) {
+ throw new AssertionError("expected " + BUFFER_SIZE + " bytes at
" + offset
+ + " but read " + read);
+ }
+ });
+ }
+ }
+
+ private BlockInputStream createLargeBlockStream() throws Exception {
+ refreshFunction = mock(Function.class);
+ BlockID blockID = new BlockID(new ContainerBlockID(1, 2));
+ checksum = new Checksum(ChecksumType.NONE, 1024);
+ chunks = new ArrayList<>(1);
+ chunkDataMap = new HashMap<>();
+ blockData = generateRandomData(SOURCE_SIZE);
+ blockSize = SOURCE_SIZE;
+ String chunkName = "large-chunk";
+ ChunkInfo chunkInfo = ChunkInfo.newBuilder()
+ .setChunkName(chunkName)
+ .setOffset(0)
+ .setLen(SOURCE_SIZE)
+ .setChecksumData(checksum.computeChecksum(
+ blockData, 0, SOURCE_SIZE).getProtoBufMessage())
+ .build();
+ chunkDataMap.put(chunkName, blockData);
+ chunks.add(chunkInfo);
+
+ OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class);
+ clientConfig.setChecksumVerify(false);
+ Pipeline pipeline = MockPipeline.createSingleNodePipeline();
+ return new DummyBlockInputStream(blockID, blockSize, pipeline, null,
+ null, refreshFunction, chunks, chunkDataMap, clientConfig);
+ }
}
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java
index 248ea865522..96d911319f0 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java
@@ -18,6 +18,8 @@
package org.apache.hadoop.hdds.scm.storage;
import static
org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.getReadChunkResponse;
+import static
org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.BUFFER_SIZE;
+import static
org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.SOURCE_SIZE;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -156,6 +158,75 @@ public void testPartialChunkRead() throws Exception {
matchWithInputData(chunkStream.getReadByteBuffers(), 0, 60);
}
+ @Test
+ public void testPositionedReadIsStateless() throws Exception {
+ // Advance the stream's own cursor with a normal sequential read.
+ byte[] seq = new byte[10];
+ assertEquals(10, chunkStream.read(seq, 0, 10));
+ matchWithInputData(seq, 0, 10);
+ long posBeforePositionedRead = chunkStream.getPos();
+
+ // A positioned read at an unrelated offset returns the right bytes ...
+ ByteBuffer dst = ByteBuffer.allocate(15);
+ int read = chunkStream.readPositioned(50, dst);
+ assertEquals(15, read);
+ dst.flip();
+ byte[] positioned = new byte[15];
+ dst.get(positioned);
+ matchWithInputData(positioned, 50, 15);
+
+ // ... without disturbing the stream's own position ...
+ assertEquals(posBeforePositionedRead, chunkStream.getPos());
+
+ // ... so the sequential read continues where it left off.
+ byte[] seq2 = new byte[10];
+ assertEquals(10, chunkStream.read(seq2, 0, 10));
+ matchWithInputData(seq2, 10, 10);
+ }
+
+ @Test
+ public void testPositionedReadAtEof() throws Exception {
+ ByteBuffer dst = ByteBuffer.allocate(10);
+ assertEquals(-1, chunkStream.readPositioned(CHUNK_SIZE, dst));
+ assertEquals(-1, chunkStream.readPositioned(CHUNK_SIZE + 1, dst));
+ }
+
+ @Test
+ public void testPositionedReadTruncatedAtChunkEnd() throws Exception {
+ // Request more than remains in the chunk; only the remainder is returned.
+ ByteBuffer dst = ByteBuffer.allocate(50);
+ int read = chunkStream.readPositioned(CHUNK_SIZE - 20, dst);
+ assertEquals(20, read);
+ dst.flip();
+ byte[] tail = new byte[20];
+ dst.get(tail);
+ matchWithInputData(tail, CHUNK_SIZE - 20, 20);
+ }
+
+ @Test
+ public void testConcurrentPositionedRead() throws Exception {
+ Checksum checksum = new Checksum(ChecksumType.CRC32, BYTES_PER_CHECKSUM);
+ byte[] largeChunkData = generateRandomData(SOURCE_SIZE);
+ ChunkInfo largeChunkInfo = ChunkInfo.newBuilder()
+ .setChunkName("large-chunk")
+ .setOffset(0)
+ .setLen(SOURCE_SIZE)
+ .setChecksumData(checksum.computeChecksum(
+ largeChunkData, 0, SOURCE_SIZE).getProtoBufMessage())
+ .build();
+ try (DummyChunkInputStream largeStream = new DummyChunkInputStream(
+ largeChunkInfo, blockID, null, false, largeChunkData.clone(), null)) {
+ PositionedReadTestHelper.runConcurrentPositionedReads(largeChunkData,
+ (offset, buf) -> {
+ int read = largeStream.readPositioned(offset, buf);
+ if (read != BUFFER_SIZE) {
+ throw new AssertionError("expected " + BUFFER_SIZE + " bytes at
" + offset
+ + " but read " + read);
+ }
+ });
+ }
+ }
+
@Test
public void testSeek() throws Exception {
seekAndVerify(0);
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestMultipartInputStream.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestMultipartInputStream.java
new file mode 100644
index 00000000000..59a010c127f
--- /dev/null
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestMultipartInputStream.java
@@ -0,0 +1,126 @@
+/*
+ * 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.hadoop.hdds.scm.storage;
+
+import static
org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.SOURCE_SIZE;
+import static
org.apache.hadoop.hdds.scm.storage.TestChunkInputStream.generateRandomData;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import com.google.common.primitives.Bytes;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.client.ContainerBlockID;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType;
+import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
+import org.apache.hadoop.hdds.scm.OzoneClientConfig;
+import org.apache.hadoop.hdds.scm.pipeline.MockPipeline;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.ozone.common.Checksum;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link MultipartInputStream}'s functionality.
+ */
+public class TestMultipartInputStream {
+
+ private static final int PART_SIZE = SOURCE_SIZE / 2;
+
+ @Test
+ public void testPositionedReadPartialAtEof() throws Exception {
+ int fileLen = 1024;
+ byte[] partData = generateRandomData(fileLen);
+ OzoneConfiguration conf = new OzoneConfiguration();
+ OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class);
+ clientConfig.setChecksumVerify(false);
+ Pipeline pipeline = MockPipeline.createSingleNodePipeline();
+ Function<BlockID, BlockLocationInfo> refreshFunction =
mock(Function.class);
+ Checksum checksum = new Checksum(ChecksumType.NONE, 1024);
+
+ BlockInputStream part = createBlockStream(new BlockID(new
ContainerBlockID(1, 1)),
+ partData, pipeline, refreshFunction, clientConfig, checksum);
+ try (MultipartInputStream multipartStream =
+ new MultipartInputStream("test-key", Collections.singletonList(part)))
{
+ multipartStream.initialize();
+ int position = 12;
+ int expectedBytes = fileLen - position;
+ ByteBuffer buffer = ByteBuffer.allocate(expectedBytes * 2);
+ assertTrue(multipartStream.readFully(position, buffer));
+ assertEquals(expectedBytes, buffer.position());
+ }
+ }
+
+ @Test
+ public void testConcurrentPositionedRead() throws Exception {
+ byte[] part0Data = generateRandomData(PART_SIZE);
+ byte[] part1Data = generateRandomData(PART_SIZE);
+ byte[] keyData = Bytes.concat(part0Data, part1Data);
+
+ OzoneConfiguration conf = new OzoneConfiguration();
+ OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class);
+ clientConfig.setChecksumVerify(false);
+ Pipeline pipeline = MockPipeline.createSingleNodePipeline();
+ Function<BlockID, BlockLocationInfo> refreshFunction =
mock(Function.class);
+ Checksum checksum = new Checksum(ChecksumType.NONE, 1024);
+
+ BlockInputStream part0 = createBlockStream(new BlockID(new
ContainerBlockID(1, 1)),
+ part0Data, pipeline, refreshFunction, clientConfig, checksum);
+ BlockInputStream part1 = createBlockStream(new BlockID(new
ContainerBlockID(1, 2)),
+ part1Data, pipeline, refreshFunction, clientConfig, checksum);
+
+ List<BlockInputStream> parts = new ArrayList<>();
+ parts.add(part0);
+ parts.add(part1);
+ try (MultipartInputStream multipartStream = new
MultipartInputStream("test-key", parts)) {
+ multipartStream.initialize();
+ PositionedReadTestHelper.runConcurrentPositionedReads(keyData,
+ (offset, buf) -> {
+ if (!multipartStream.readFully(offset, buf)) {
+ throw new AssertionError("stateless readFully returned false at
" + offset);
+ }
+ });
+ }
+ }
+
+ private static BlockInputStream createBlockStream(BlockID blockID, byte[]
blockData,
+ Pipeline pipeline, Function<BlockID, BlockLocationInfo> refreshFunction,
+ OzoneClientConfig clientConfig, Checksum checksum) throws Exception {
+ List<ChunkInfo> chunks = new ArrayList<>(1);
+ Map<String, byte[]> chunkDataMap = new HashMap<>();
+ String chunkName = "chunk-" + blockID.getLocalID();
+ ChunkInfo chunkInfo = ChunkInfo.newBuilder()
+ .setChunkName(chunkName)
+ .setOffset(0)
+ .setLen(blockData.length)
+ .setChecksumData(checksum.computeChecksum(
+ blockData, 0, blockData.length).getProtoBufMessage())
+ .build();
+ chunkDataMap.put(chunkName, blockData);
+ chunks.add(chunkInfo);
+ return new DummyBlockInputStream(blockID, blockData.length, pipeline, null,
+ null, refreshFunction, chunks, chunkDataMap, clientConfig);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
