szetszwo commented on PR #11102:
URL: https://github.com/apache/ozone/pull/11102#issuecomment-5642450252
> the tests with DummyChunkInputStream and the actual use case of
LocalChunkInputStream may fail with NPE for datanodeBlockID if we change this
back to private
Overriding the acquireClient/readChunk makes code confusing; see below for
my suggestion
> unless we agreed that the shared xceiverClient will not be set to null
during the readPositioned(chunkRelativePosition, dst) when the releaseClient
is called by either sequential read's handleReadError or unbuffer. if so, we
can revert this readChunk(ChunkInfo, XceiverClientSpi) and have a simple
readChunk(ChunkInfo) like before this change.
- You are right that sequential read can release the client. Then, even if
the client is non-null in pos read, it is still incorrect to use it after
release.
- BTW, the XceiverClientSpi implementations are not thread safe. Using the
same client for concurrent pos reads is also incorrect.
So, I suggest to acquire a client directly from the factory and release it
after use:
```java
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) {
return copyRange(readChunk(readChunkInfo), skip, toRead, dst);
}
final Pipeline pipeline = pipelineSupplier.get();
final ContainerProtos.DatanodeBlockID bid =
buildDatanodeBlockId(pipeline, blockID);
final XceiverClientSpi client =
xceiverClientFactory.acquireClientForReadData(pipeline);
try {
final ByteBuffer[] readBuffers = readChunk(client, readChunkInfo, bid,
validators, tokenSupplier.get());
return copyRange(readBuffers, skip, toRead, dst);
} finally {
xceiverClientFactory.releaseClientForReadData(client, false);
}
}
```
```java
private static ByteBuffer[] readChunk(
XceiverClientSpi client, ChunkInfo chunk, 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()
.toArray(new ByteBuffer[0]);
} else if (readChunkResponse.hasDataBuffers()) {
List<ByteString> buffersList = readChunkResponse.getDataBuffers()
.getBuffersList();
return BufferUtils.getReadOnlyByteBuffersArray(buffersList);
} else {
throw new IOException("Unexpected error while reading chunk data " +
"from container. No data returned.");
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
