Copilot commented on code in PR #11348:
URL: https://github.com/apache/ozone/pull/11348#discussion_r4122351635
##########
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java:
##########
@@ -66,6 +70,28 @@ public int read(ByteBuffer byteBuffer) throws IOException {
}
}
+ @Override
+ protected int readWithStrategy(ByteReaderStrategy strategy) throws
IOException {
+ return strategy.readFromBlock(inputStream, strategy.getTargetLength());
+ }
+
+ @Override
+ protected int readPositioned(long position, ByteBuffer buffer) throws
IOException {
+ if (inputStream instanceof ByteBufferPositionedReadable) {
+ return ((ByteBufferPositionedReadable) inputStream).read(position,
buffer);
+ }
+ throw new UnsupportedOperationException("Positioned reads are not
supported by "
+ + inputStream.getClass().getName());
+ }
+
+ @Override
+ public boolean hasCapability(String capability) {
+ if (StreamCapabilities.PREADBYTEBUFFER.equalsIgnoreCase(capability)) {
+ return inputStream instanceof ByteBufferPositionedReadable;
+ }
+ return super.hasCapability(capability);
Review Comment:
This wrapper now advertises capabilities that its delegates may not actually
provide. `ExtendedInputStream.hasCapability` returns `true` for
`READBYTEBUFFER` and `UNBUFFER`, but `read(ByteBuffer)` only works when the
wrapped stream implements `ByteBufferReadable`; wrapping a plain
`ByteArrayInputStream` therefore reports support and then throws. The
`PREADBYTEBUFFER` branch also ignores an underlying `StreamCapabilities`
implementation that explicitly reports false (as the cursor-only EC streams
do), so mirror the delegate checks for all three capabilities instead of
falling back to `super`.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java:
##########
@@ -178,6 +181,144 @@ synchronized int readFully(ByteBuffer targetBuf, boolean
preRead) throws IOExcep
return read > 0 ? read : EOF;
}
+ /**
+ * Positioned read served by a dedicated one-shot streaming read which never
touches the
+ * cursor, the pre-read buffer or the sequential reader.
+ *
+ * @param blockOffset the offset within the block to read from.
+ * @param dst the buffer to read into.
+ * @return the number of bytes copied into {@code dst}, or -1 if no byte
could be read.
+ */
+ @Override
+ protected int readPositioned(long blockOffset, ByteBuffer dst) throws
IOException {
+ if (!dst.hasRemaining()) {
+ return 0;
+ }
+ if (blockOffset < 0 || blockOffset >= blockLength) {
+ return EOF;
+ }
+ final XceiverClientFactory factory;
+ synchronized (this) {
+ checkOpen();
+ factory = xceiverClientFactory;
+ }
+ final int length = Math.toIntExact(Math.min(dst.remaining(), blockLength -
blockOffset));
+ final int startPosition = dst.position();
+ int callRetries = 0;
+ while (true) {
+ final AtomicReference<OneShotReader> readerRef = new AtomicReference<>();
+ try {
+ return preadOnce(factory, readerRef, blockOffset, length, dst);
+ } catch (IOException e) {
+ dst.position(startPosition);
+ handlePreadException(e, readerRef.get(), blockOffset, callRetries++);
+ }
+ }
+ }
+
+ private int preadOnce(XceiverClientFactory factory,
AtomicReference<OneShotReader> readerRef,
+ long blockOffset, int length, ByteBuffer dst) throws IOException {
+ final Pipeline pipeline = pipelineRef.get();
+ final XceiverClientSpi acquired =
factory.acquireClientForReadData(pipeline);
+ if (acquired == null) {
+ throw new IOException("Failed to acquire client for " + pipeline);
+ }
+ if (!(acquired instanceof XceiverClientGrpc)) {
+ throw new IOException("Unexpected client class: " +
acquired.getClass().getName() + ", " + pipeline);
Review Comment:
When the factory returns a non-gRPC `XceiverClientSpi`, this branch throws
after acquiring the client but before the `finally` that calls
`releaseClientForReadData(client, false)`. That leaks the read client for this
positioned-read attempt; move the type check inside a `try/finally` that
releases the original `XceiverClientSpi` (while retaining the existing
gRPC-specific cleanup).
--
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]
