[FLINK-20654][checkpoint] Only trace buffers that are actually persis…

…ted.
deepfield · Mar 26, 2021 · dcd40e9 · dcd40e9
1 parent cb8113a
commit dcd40e9
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 7 deletions.
diff --git a/...c/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateCheckpointWriter.java b/...c/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateCheckpointWriter.java
@@ -167,6 +167,7 @@ private <K> void write(
  long size = checkpointStream.getPos() - offset;
  offsets.computeIfAbsent(key, unused -> new StateContentMetaInfo())
  .withDataAdded(offset, size);
+ NetworkActionsLogger.tracePersist(action, buffer, key, checkpointId);
  });
  } finally {
  buffer.recycleBuffer();

diff --git a/...e/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequest.java b/...e/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequest.java
@@ -19,7 +19,6 @@
 
 import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter.ChannelStateWriteResult;
 import org.apache.flink.runtime.io.network.buffer.Buffer;
-import org.apache.flink.runtime.io.network.logger.NetworkActionsLogger;
 import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
 import org.apache.flink.util.CloseableIterator;
 import org.apache.flink.util.Preconditions;
@@ -59,8 +58,7 @@ static ChannelStateWriteRequest write(
  long checkpointId, InputChannelInfo info, CloseableIterator<Buffer> iterator) {
  return buildWriteRequest(
  checkpointId,
- "ChannelStateWriteRequest#writeInput",
- info,
+ "writeInput",
  iterator,
  (writer, buffer) -> writer.writeInput(info, buffer));
  }
@@ -69,16 +67,14 @@ static ChannelStateWriteRequest write(
  long checkpointId, ResultSubpartitionInfo info, Buffer... buffers) {
  return buildWriteRequest(
  checkpointId,
- "ChannelStateWriteRequest#writeOutput",
- info,
+ "writeOutput",
  ofElements(Buffer::recycleBuffer, buffers),
  (writer, buffer) -> writer.writeOutput(info, buffer));
  }
 
  static ChannelStateWriteRequest buildWriteRequest(
  long checkpointId,
  String name,
- Object channelInfo,
  CloseableIterator<Buffer> iterator,
  BiConsumerWithException<ChannelStateCheckpointWriter, Buffer, Exception>
  bufferConsumer) {
@@ -88,7 +84,6 @@ static ChannelStateWriteRequest buildWriteRequest(
  writer -> {
  while (iterator.hasNext()) {
  Buffer buffer = iterator.next();
- NetworkActionsLogger.tracePersist(name, buffer, channelInfo, checkpointId);
  try {
  checkArgument(buffer.isBuffer());
  } catch (Exception e) {