peterxcli commented on code in PR #11348:
URL: https://github.com/apache/ozone/pull/11348#discussion_r4191534327
##########
hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java:
##########
@@ -193,13 +188,23 @@ public int read(long position, byte[] array, int offset,
int length) throws IOEx
return readImpl(position, ByteBuffer.wrap(array, offset, length));
}
+ protected InputStream getWrappedInputStream() {
+ return inputStream;
+ }
+
private int readImpl(long position, ByteBuffer buffer) throws IOException {
Review Comment:
The old fallback casts to `Seekable`, which `CipherInputStream` doesn't
implement, so GDPR positioned reads already failed. Removing support for other
seek-only delegates is intentional and covered by a test.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java:
##########
@@ -100,6 +128,7 @@ public synchronized boolean seekToNewSource(long l) throws
IOException {
@Override
public boolean hasCapability(String capability) {
Review Comment:
The proposed hook still defaults to true, so unsupported streams would still
need to opt out. I'd keep the existing overrides.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java:
##########
@@ -195,45 +180,35 @@ public synchronized void seek(long pos) throws
IOException {
}
@Override
- public boolean readFully(long position, ByteBuffer buffer) throws
IOException {
- if (!statelessPositionedReadSupported) {
- return false;
- }
- return readFullyStateless(position, buffer);
- }
-
- /**
- * Shared stateless positioned-read loop across parts using {@link
PartInputStream#readPositioned}.
- */
- private boolean readFullyStateless(long position, ByteBuffer buffer) throws
IOException {
+ protected int readPositioned(long position, ByteBuffer buffer) throws
IOException {
+ checkOpen();
if (!buffer.hasRemaining()) {
- return true;
+ return 0;
}
-
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);
+ while (buffer.hasRemaining() && pos < length) {
+ int index = binarySearchOffsetIndex(partOffsets, pos);
+ PartInputStream part = partStreams.get(index);
+ long partPosition = pos - partOffsets[index];
+ final int oldLimit = buffer.limit();
+ final int n;
+ try {
+ buffer.limit(buffer.position() + (int) Math.min(buffer.remaining(),
part.getLength() - partPosition));
+ n = part.read(partPosition, buffer);
+ } finally {
+ buffer.limit(oldLimit);
+ }
+ if (n < 0) {
+ break;
}
- int idx = binarySearchOffsetIndex(partOffsets, pos);
- long partPos = pos - partOffsets[idx];
- int n = partStreams.get(idx).readPositioned(partPos, buffer);
- if (n <= 0) {
- if (bytesRead > 0) {
- return true;
- }
- throw new EOFException("EOF encountered at pos: " + pos +
- " for key: " + key);
+ if (n == 0) {
Review Comment:
The write paths omit zero-length blocks when committing to OM, so we don't
need an empty-part guard here. I've removed the buffer clamp and left bounds
handling to each part.
##########
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java:
##########
@@ -66,6 +70,38 @@ 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 {
Review Comment:
RpcClient always supplies a stream, and I found no production callers of the
no-arg constructor. I'd leave null handling out of this PR.
##########
hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/CapableOzoneFSInputStream.java:
##########
@@ -34,8 +35,9 @@ public boolean hasCapability(String capability) {
switch (StringUtils.toLowerCase(capability)) {
case StreamCapabilities.READBYTEBUFFER:
case StreamCapabilities.UNBUFFER:
- case StreamCapabilities.PREADBYTEBUFFER:
return true;
+ case StreamCapabilities.PREADBYTEBUFFER:
Review Comment:
Both filesystem adapters call `bucket.readFile(key).getInputStream()`, which
unwraps `OzoneInputStream`. For GDPR keys, the delegate is the raw
`CipherInputStream`, so this check returns false.
##########
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/OzoneInputStream.java:
##########
@@ -66,6 +70,38 @@ public int read(ByteBuffer byteBuffer) throws IOException {
}
}
+ @Override
+ protected int readWithStrategy(ByteReaderStrategy strategy) throws
IOException {
Review Comment:
The inherited public `read(ByteReaderStrategy)` calls this method, so it is
reachable. Making it throw would break that API.
--
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]
