Github user NicoK commented on a diff in the pull request:
https://github.com/apache/flink/pull/4581#discussion_r152841030
--- 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 --
Actually, a sanity check for double-recycle comes with #4613 for which I
also needed this PR. It does, however, work differently and only checks that
the reference counter does not go below 0 - I guess, this way we do not put too
much pressure on the garbage collector compared to creating new Buffer
instances for each `retain()`
---
