Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/4581#discussion_r152825430
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java
---
@@ -267,28 +268,29 @@ public ResultPartitionType getPartitionType() {
* first buffer has been added.
*/
public void add(Buffer buffer, int subpartitionIndex) throws
IOException {
- boolean success = false;
+ checkNotNull(buffer);
try {
checkInProduceState();
+ } catch (Throwable t) {
+ buffer.recycle();
--- End diff --
I wonder if we should have some sanity illegal state detection for double
recycling the buffers. For example each buffer could only be recycled once
(protected by a private field in the Buffer `boolean wasRecycled`). Whenever
you call `retain()`, you would get a new instance of the `Buffer`, pointing to
the same memory, but with new flag (so that both original and retained buffers
could be recycled independently, but each one of them only once).
---
