zhijiangW 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_r355312789
##########
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:
Actually this should be a separate bug out of this ticket scope, so it is
better to submit this fix separately.
In addition this issue let me think why it was not exposed by tests before.
I think the following `bufferConsumer.close()` somehow allows the inconsistent
logics to bring potential risks. In implementing of `BufferConsumer#close()`
it would firstly check whether the backed buffer was released or not. If not
released yet, then it would call `Buffer#recycleBuffer`, otherwise it does
nothing.
I am not sure why we define such loose semantics for `BufferConsumer#close`,
rather than make it as a strict semantic. Do you have any inputs here
@pnowojski ?
----------------------------------------------------------------
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
