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 2025/08/28 17:50:47 UTC
(flink) branch master updated: [FLINK-38288] Enhance BlobWriter logging with offloading time in millis.
This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 280767445c1 [FLINK-38288] Enhance BlobWriter logging with offloading time in millis.
280767445c1 is described below
commit 280767445c17db090bfc65e0e8d09374b9ea92bf
Author: Aleksandr Iushmanov <ai...@confluent.io>
AuthorDate: Wed Aug 27 17:31:52 2025 +0100
[FLINK-38288] Enhance BlobWriter logging with offloading time in millis.
---
.../org/apache/flink/runtime/blob/BlobWriter.java | 19 +++++++++++++++++--
1 file changed, 17 insertions(+), 2 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobWriter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobWriter.java
index 2d5292b42cb..d7b7c542363 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobWriter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobWriter.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.types.Either;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedValue;
+import org.apache.flink.util.clock.SystemClock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -111,12 +112,26 @@ public interface BlobWriter {
Preconditions.checkNotNull(serializedValue);
Preconditions.checkNotNull(jobId);
Preconditions.checkNotNull(blobWriter);
+ final SystemClock systemClock = SystemClock.getInstance();
+ final long offloadStartTs = systemClock.relativeTimeMillis();
try {
final PermanentBlobKey permanentBlobKey =
blobWriter.putPermanent(jobId, serializedValue.getByteArray());
- return Either.Right(permanentBlobKey);
+ final Either<SerializedValue<T>, PermanentBlobKey> right =
+ Either.Right(permanentBlobKey);
+ LOG.info(
+ "BLOB for job {} with size: {} has been offloaded in {} millis",
+ jobId,
+ serializedValue.getByteArray().length,
+ systemClock.relativeTimeMillis() - offloadStartTs);
+ return right;
} catch (IOException e) {
- LOG.warn("Failed to offload value for job {} to BLOB store.", jobId, e);
+ LOG.warn(
+ "Failed to offload value for job {} to BLOB store with size: {} after {} millis",
+ jobId,
+ serializedValue.getByteArray().length,
+ systemClock.relativeTimeMillis() - offloadStartTs,
+ e);
return Either.Left(serializedValue);
}
}