Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobWriter.java (Java)

Code before:
        Preconditions.checkNotNull(blobWriter);

        if (serializedValue.getByteArray().length < blobWriter.getMinOffloadingSize()) {
            return Either.Left(serializedValue);
        } else {
            return offloadWithException(serializedValue, jobId, blobWriter);
        }
    }

    static <T> Either<SerializedValue<T>, PermanentBlobKey> offloadWithException(
            SerializedValue<T> serializedValue, JobID jobId, BlobWriter blobWriter) {
        Preconditions.checkNotNull(serializedValue);
        Preconditions.checkNotNull(jobId);
        Preconditions.checkNotNull(blobWriter);
        LOG.info(
                "Offloading BLOB for job {}, size: {}",
                jobId,
                serializedValue.getByteArray().length);
        final SystemClock systemClock = SystemClock.getInstance();
        final long offloadStartTs = systemClock.relativeTimeMillis();
        try {
            final PermanentBlobKey permanentBlobKey =
                    blobWriter.putPermanent(jobId, serializedValue.getByteArray());
            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 after {} millis",
                    jobId,
                    systemClock.relativeTimeMillis() - offloadStartTs,
                    e);
            return Either.Left(serializedValue);
        }
    }
}


Review comment (suggestion):
This PR introduced 3 logs. If we have the second log, whether the first one is not needed? 

- The first log has been included by the second one, right?
- Even if it throws some exceptions during offloading, the third log works. 

If removing the first log makes sense, it would be better to log `serializedValue.getByteArray().length` in the third log.

Code after:
        Preconditions.checkNotNull(blobWriter);

        if (serializedValue.getByteArray().length < blobWriter.getMinOffloadingSize()) {
            return Either.Left(serializedValue);
        } else {
            return offloadWithException(serializedValue, jobId, blobWriter);
        }
    }

    static <T> Either<SerializedValue<T>, PermanentBlobKey> offloadWithException(
            SerializedValue<T> serializedValue, JobID jobId, BlobWriter 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());
            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 with size: {} after {} millis",
                    jobId,
                    serializedValue.getByteArray().length,
                    systemClock.relativeTimeMillis() - offloadStartTs,
                    e);
            return Either.Left(serializedValue);
        }
    }
}