Zakelly commented on code in PR #25509:
URL: https://github.com/apache/flink/pull/25509#discussion_r1805752878
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataInputStream.java:
##########
@@ -31,19 +31,21 @@ public class CompressibleFSDataInputStream extends
FSDataInputStream {
private final FSDataInputStream delegate;
private final InputStream compressingDelegate;
+ private final boolean compressed;
public CompressibleFSDataInputStream(
FSDataInputStream delegate, StreamCompressionDecorator
compressionDecorator)
throws IOException {
this.delegate = delegate;
this.compressingDelegate =
compressionDecorator.decorateWithCompression(delegate);
+ this.compressed = compressionDecorator !=
UncompressedStreamCompressionDecorator.INSTANCE;
}
@Override
public void seek(long desired) throws IOException {
- final int available = compressingDelegate.available();
- if (available > 0) {
- if (available != compressingDelegate.skip(available)) {
+ if (compressed) {
+ final int available = compressingDelegate.available();
+ if (available > 0 && available !=
compressingDelegate.skip(available)) {
throw new IOException("Unable to skip buffered data.");
}
}
Review Comment:
Separate classes uncompressed and compressed `FSDataInputStream` sounds good
to me. Agree to consider this later.
--
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]
