KAFKA-19893: Reduce tiered storage redundancy with delayed upload (KIP-1241) #20913
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -66,6 +66,7 @@ | |
| import org.apache.kafka.storage.internals.log.AsyncOffsetReader; | ||
| import org.apache.kafka.storage.internals.log.EpochEntry; | ||
| import org.apache.kafka.storage.internals.log.FetchDataInfo; | ||
| import org.apache.kafka.storage.internals.log.LogConfig; | ||
| import org.apache.kafka.storage.internals.log.LogOffsetMetadata; | ||
| import org.apache.kafka.storage.internals.log.LogSegment; | ||
| import org.apache.kafka.storage.internals.log.OffsetIndex; | ||
|
|
@@ -916,6 +917,7 @@ private void maybeUpdateCopiedOffset(UnifiedLog log) throws RemoteStorageExcepti | |
| * 1) Segment is not the active segment and | ||
| * 2) Segment end-offset is less than the last-stable-offset as remote storage should contain only | ||
| * committed/acked messages | ||
| * 3) Segment has exceeded copy lag by time or size when configured (remote.copy.lag.ms, remote.copy.lag.bytes) | ||
| * @param log The log from which the segments are to be copied | ||
| * @param fromOffset The offset from which the segments are to be copied | ||
| * @param lastStableOffset The last stable offset of the log | ||
|
|
@@ -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) { | ||
| 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) { | ||
|
Member
There was a problem hiding this comment. Should we log a warning or info message when a user configures one lag property but leaves the other at its default value of 0?
Contributor
Author
There was a problem hiding this comment. @chia7712 Thanks for your view.
Member
There was a problem hiding this comment. Sounds good. We can expand |
||
| 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; | ||
|
Member
There was a problem hiding this comment. This is a lingering issue. The local segment with a future timestamp is still NOT deleted, right? Should we allow the deletion of these local files once they have been successfully uploaded to remote storage, even if they contain future timestamps?
Contributor
There was a problem hiding this comment.
Currently, it gets deleted based on the local retention bytes limit since the local retention deletion logic follow the similar methods of full deletion logic.
Member
There was a problem hiding this comment. The current behavior looks like a bug to me, as local segments shouldn't be blocked from deletion once they are already uploaded. Perhaps we could fall back to checking the lastModifiedTime when a segment contains future records. Alternatively, we could introduce a new configuration flag to provide alternative logic for handling future records.
Contributor
There was a problem hiding this comment.
Yeah, we can allow this behavior by introducing a config. The segment exist in the remote storage but the user might face slowness in reading the data from remote if they don't have prefetching feature implemented in the Remote Storage Manager. So, better to gate the change in behavior via config and the change applies only when remote storage is enabled on the topic.
Member
There was a problem hiding this comment. If we all hate For example, we could introduce remote.copy.before.deletion.ms instead of remote.copy.lag.ms. If a user sets remote.copy.before.deletion.ms=2hr, it means they want to upload log segments 2 hours before they are scheduled to be deleted from local storage. The default value would be -1, which dynamically resolves to local.retention.ms. This fully maintains backward compatibility (resulting in immediate upload). Setting it to 0 means the user wants to delay the upload as much as possible, triggering it right before local deletion. The only side effect is that this kind of "reverse" or "countdown" time calculation is quite rare in the existing Kafka codebase.
Contributor
There was a problem hiding this comment.
yeah, this is not intuitive when compared with the other configs like retention time/bytes. My suggestion is to:
Member
There was a problem hiding this comment. I just realized that I replied to the wrong thread earlier, sorry about that 😢
Yes, that is an acceptable approach. I've opened https://issues.apache.org/jira/browse/KAFKA-20609 so we can keep discussing there.
While documentation is the final line of defense for users, it's better to have an intuitive design out of the box. After all, me proposed reverse configuration was rejected precisely because it wasn't intuitive enough Another way is to align the logic with retention.ms and retention.bytes. Since most users care more about time than size, we could set the default value of the time lag to 0, and the size lag to -1. This way, most users can just adjust the time setting without having to touch the size configuration. WDYT?
Contributor
There was a problem hiding this comment.
I like the idea to provide out-of-box default values that works in majority of use-cases and retain the eager upload logic. The new default values for remote-copy lag align with the
Contributor
Author
|
||
| if (logger.isTraceEnabled()) { | ||
| logger.trace("{} eligible for upload by time? {} (segment age {} ms, copy lag {} ms)", | ||
| segment, eligibleUpload, segmentAgeMs, copyLagMs); | ||
| } | ||
| return eligibleUpload; | ||
| } catch (IOException e) { | ||
| logger.warn("Failed to get largest timestamp for segment {}, take it as eligible for upload based on time", segment, e); | ||
| return true; | ||
| } | ||
| } | ||
|
|
||
| private boolean eligibleUploadBySize(LogSegment segment, long totalLogSize, long cumulativeSize, long copyLagBytes) { | ||
| long sizeLagBytes = totalLogSize - cumulativeSize; | ||
| boolean eligibleUpload = sizeLagBytes >= copyLagBytes; | ||
| if (logger.isTraceEnabled()) { | ||
| logger.trace("{} eligible for upload by size? {} (size lag {} bytes, copy lag {} bytes, totalLogSize={}, cumulativeSize={})", | ||
| segment, eligibleUpload, sizeLagBytes, copyLagBytes, totalLogSize, cumulativeSize); | ||
| } | ||
| return eligibleUpload; | ||
| } | ||
|
|
||
| public void copyLogSegmentsToRemote(UnifiedLog log) throws InterruptedException, RetriableRemoteStorageException { | ||
| if (isCancelled()) | ||
| return; | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
could you also add one test for valid dynamic broker config change? Thanks!
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Also, move the
testDynamicRemoteCopyLagThrowsOnIncorrectConfigtest to DynamicBrokerConfigTest.java instead of DynamicBrokerConfigTest.scala.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This can be taken up later, we already have one dynamic broker config change test in KafkaConfigTest.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Got it. thanks @kamalcph
Thank you very much for your patient and thorough review.