jiafu1115 commented on code in PR #20913:
URL: https://github.com/apache/kafka/pull/20913#discussion_r3295457433
##########
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java:
##########
@@ -925,18 +927,73 @@ List<EnrichedLogSegment> candidateLogSegments(UnifiedLog
log, Long fromOffset, L
List<EnrichedLogSegment> candidateLogSegments = new ArrayList<>();
List<LogSegment> segments = log.logSegments(fromOffset,
Long.MAX_VALUE);
if (!segments.isEmpty()) {
+ long currentTimeMs = time.milliseconds();
+ long totalLogSize = UnifiedLog.sizeInBytes(segments);
+ long cumulativeSize = 0;
for (int idx = 1; idx < segments.size(); idx++) {
LogSegment previousSeg = segments.get(idx - 1);
LogSegment currentSeg = segments.get(idx);
if (currentSeg.baseOffset() <= lastStableOffset) {
- candidateLogSegments.add(new
EnrichedLogSegment(previousSeg, currentSeg.baseOffset()));
+ cumulativeSize += previousSeg.size();
+ if (isEligibleForUpload(log.config(), previousSeg,
currentTimeMs, totalLogSize, cumulativeSize)) {
+ candidateLogSegments.add(new
EnrichedLogSegment(previousSeg, currentSeg.baseOffset()));
+ } else {
+ break;
+ }
}
}
// Discard the last active segment
}
return candidateLogSegments;
}
+ private boolean isEligibleForUpload(LogConfig logConfig, LogSegment
previousSeg, long currentTimeMs, long totalLogSize, long cumulativeSize) {
+ long copyLagMs = logConfig.remoteCopyLagMs();
+ long copyLagBytes = logConfig.remoteCopyLagBytes();
+ if (logger.isTraceEnabled()) {
+ logger.trace("delayCopy check for segment {}: copyLagMs={},
copyLagBytes={}, currentTimeMs={}, totalLogSize={}, cumulativeSize={},
sizeLagBytes={}",
+ previousSeg, copyLagMs, copyLagBytes, currentTimeMs,
totalLogSize, cumulativeSize, totalLogSize - cumulativeSize);
+ }
+
+ if (copyLagMs == 0 || copyLagBytes == 0) {
+ return true;
+ }
+
+ boolean limitedCopyLagMsCheck = copyLagMs > 0;
+ boolean limitedCopyLagSizeCheck = copyLagBytes > 0;
+
+ if (limitedCopyLagMsCheck && eligibleUploadByTime(previousSeg,
currentTimeMs, copyLagMs)) {
+ return true;
+ }
+
+ return limitedCopyLagSizeCheck &&
eligibleUploadBySize(previousSeg, totalLogSize, cumulativeSize, copyLagBytes);
+ }
+
+ private boolean eligibleUploadByTime(LogSegment segment, long
currentTimeMs, long copyLagMs) {
+ try {
+ long segmentAgeMs = currentTimeMs - segment.largestTimestamp();
+ boolean eligibleUpload = segmentAgeMs < 0 || segmentAgeMs >=
copyLagMs;
Review Comment:
@kamalcph @chia7712 thanks for raising so interesting topic (future
timestamp). Let me try to undersand it.
For local segment of this case. It and the segements after it won't be
deleted due to follow logic:
<img width="1136" height="186" alt="image"
src="https://github.com/user-attachments/assets/82106771-16b5-4eae-9dfe-0df5a9cdc570";
/>
So, local segments will not be deleted regardless of whether they have been
uploaded to remote storage, so there is no performance concern here. In the
worst case, we simply achieve less cost reduction for this scenario.
In practice, if the goal is to reduce max possible costs while preserving
the existing behavior, delaying the upload would also work. However, I am
concerned that the local deletion logic may evolve in the future. For that
reason, I though allowing the upload is more reasonable for this special case.
And we may have some extra solution for this case:
(1) We can document this limitation clearly and define the basic principle
of this KIP as: we aim to reduce storage costs, although this special case may
not fully benefit from the optimization.
(2) How about this solution? if we found it contain future record. we use
LogSegment#lastModified to compare the time?
(3) Delay the upload to reach the max cost saving no consider the local
delete logic for this case changed in future.
And. for the proposed solution: " remote.copy.before.deletion.ms". it seems
a little more complex to be understand than the existed solution. I am worring
if worth to revise the whole logic for this special case. But I can take more
time to consider it and give feedback before Kafka 4.4.0's code freeze.
BTW, For my proposed three solutions. WDYT?
Thanks @chia7712 @kamalcph
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
