Zakelly commented on code in PR #25509:
URL: https://github.com/apache/flink/pull/25509#discussion_r1804708194
##########
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:
How about moving this part into a new interface. For example,
`compressingDelegate` is a newly introduced `DecompressingInputStream`, which
provides a method `clearBuffer`. Here's where the different behaviors go.
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
