You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fa...@apache.org on 2026/07/09 08:11:32 UTC
(flink) 05/05: [FLINK-39520][network] Additive logging overload; widen releaseAllResources visibility
This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8cd1c0ec6579b5d6efcc5d876aec1d1b5047921e
Author: Rui Fan <19...@gmail.com>
AuthorDate: Mon Jul 6 00:04:41 2026 +0200
[FLINK-39520][network] Additive logging overload; widen releaseAllResources visibility
- NetworkActionsLogger: additive tracePersist(String, Object, Object, long)
overload; the existing Buffer overload delegates to it.
- RecoveredInputChannel#releaseAllResources: package-private -> public.
---
.../runtime/io/network/logger/NetworkActionsLogger.java | 12 ++++++------
.../io/network/partition/consumer/RecoveredInputChannel.java | 2 +-
2 files changed, 7 insertions(+), 7 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/logger/NetworkActionsLogger.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/logger/NetworkActionsLogger.java
index 204f3d3c074..137334ae14f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/logger/NetworkActionsLogger.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/logger/NetworkActionsLogger.java
@@ -98,13 +98,13 @@ public class NetworkActionsLogger {
public static void tracePersist(
String action, Buffer buffer, Object channelInfo, long checkpointId) {
+ tracePersist(action, buffer.toDebugString(INCLUDE_HASH), channelInfo, checkpointId);
+ }
+
+ public static void tracePersist(
+ String action, Object persisted, Object channelInfo, long checkpointId) {
if (LOG.isTraceEnabled()) {
- LOG.trace(
- "{} {}, checkpoint {} @ {}",
- action,
- buffer.toDebugString(INCLUDE_HASH),
- checkpointId,
- channelInfo);
+ LOG.trace("{} {}, checkpoint {} @ {}", action, persisted, checkpointId, channelInfo);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java
index 168901a977b..b68ae24c8c5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java
@@ -304,7 +304,7 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan
}
}
- void releaseAllResources() throws IOException {
+ public void releaseAllResources() throws IOException {
ArrayDeque<Buffer> releasedBuffers = new ArrayDeque<>();
boolean shouldRelease = false;