Rui Fan created FLINK-40345:
-------------------------------
Summary: Buffer is recycled twice in
FilteringHandler#recoverWithFiltering when delivery fails
Key: FLINK-40345
URL: https://issues.apache.org/jira/browse/FLINK-40345
Project: Flink
Issue Type: Sub-task
Affects Versions: 2.4.0
Reporter: Rui Fan
Assignee: Rui Fan
Fix For: 2.4.0
In FilteringHandler#recoverWithFiltering, the catch block recycles buffers
starting
at index i:
{code:java}
} catch (Throwable t) {
for (int j = i; j < filteredBuffers.size(); j++) {
filteredBuffers.get(j).recycleBuffer();
}
throw t;
}{code}
But onRecoveredStateBuffer never leaves the buffer un-owned when it throws: it
enqueues the buffer before the only statement that can throw
(notifyChannelNonEmpty()). So the buffer at index i is recycled twice,
corrupting
the reference count and the buffer pool. The loop should start at i + 1.
Introduced by FLINK-38930, master only. Reported by Roman in
[https://github.com/apache/flink/pull/28661#pullrequestreview-4832855136]
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
