You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2023/03/13 15:58:27 UTC
[incubator-celeborn] branch main updated: [CELEBORN-417][FLINK] fix memory leak when handler already removed (#1342)
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new b499c0df7 [CELEBORN-417][FLINK] fix memory leak when handler already removed (#1342)
b499c0df7 is described below
commit b499c0df7f8ba981c6c0bf8fb2cc66ca904f1f69
Author: Shuang <lv...@gmail.com>
AuthorDate: Mon Mar 13 23:58:21 2023 +0800
[CELEBORN-417][FLINK] fix memory leak when handler already removed (#1342)
---
.../org/apache/celeborn/plugin/flink/network/ReadClientHandler.java | 4 ++++
.../org/apache/celeborn/plugin/flink/RemoteBufferStreamReader.java | 4 ----
2 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/network/ReadClientHandler.java b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/network/ReadClientHandler.java
index 4a7d99d3c..ed6a87015 100644
--- a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/network/ReadClientHandler.java
+++ b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/network/ReadClientHandler.java
@@ -58,6 +58,10 @@ public class ReadClientHandler extends BaseMessageHandler {
logger.debug("received streamId: {}, msg :{}", streamId, msg);
handler.accept(msg);
} else {
+ if (msg != null && msg instanceof ReadData) {
+ ((ReadData) msg).getFlinkBuffer().release();
+ }
+
logger.warn("Unexpected streamId received: {}", streamId);
}
}
diff --git a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteBufferStreamReader.java b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteBufferStreamReader.java
index 9313ff701..08c0d1fdf 100644
--- a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteBufferStreamReader.java
+++ b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteBufferStreamReader.java
@@ -140,10 +140,6 @@ public class RemoteBufferStreamReader extends CreditListener {
"Rss buffer stream reader get streamid {} received readable bytes {}.",
readData.getStreamId(),
readData.getFlinkBuffer().readableBytes());
- if (closed) {
- readData.body().release();
- return;
- }
dataListener.accept(readData.getFlinkBuffer());
}
}