pnowojski commented on a change in pull request #10492: [FLINK-15140][runtime]
Fix shuffle data compression doesn't work with BroadcastRecordWriter.
URL: https://github.com/apache/flink/pull/10492#discussion_r355327159
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/BoundedBlockingSubpartition.java
##########
@@ -152,7 +152,10 @@ private void writeAndCloseBufferConsumer(BufferConsumer
bufferConsumer) throws I
if (canBeCompressed(buffer)) {
final Buffer compressedBuffer =
parent.bufferCompressor.compressToIntermediateBuffer(buffer);
data.writeBuffer(compressedBuffer);
- compressedBuffer.recycleBuffer();
+ // release the intermediate buffer
+ if (compressedBuffer != buffer) {
+
compressedBuffer.recycleBuffer();
Review comment:
+1 this must be an independent bug fix/jira ticket, with it's own test
coverage.
Wasn't it discovered because this is a rare case, that compression couldn't
reduce the size of the data?
> I think the following bufferConsumer.close() somehow allows the
inconsistent logics to bring potential risks.
What do you mean?
> I am not sure why we define such loose semantics for BufferConsumer#close,
rather than make it as a strict semantic.
The if check might have been added to implement `Closeable` interface. Note
the difference between `Closeable` and `AutoCloseable`:
```
public interface Closeable extends AutoCloseable {
/**
* Closes this stream and releases any system resources associated
* with it. If the stream is already closed then invoking this
* method has no effect.
```
`Closeable` is an easier interface to use, as user can just do not care if
it was closed before or not. I'm not sure if we strictly need this though.
Maybe we do or maybe I've implemented it that way because I prefer `Closeable`
more ;) ).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
