chia7712 commented on code in PR #20913:
URL: https://github.com/apache/kafka/pull/20913#discussion_r3289589447
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java:
##########
@@ -608,6 +631,32 @@ private static void
validateRemoteStorageRetentionTime(Map<String, ?> props) {
}
}
+ private static void validateRemoteCopyLagTime(Map<?, ?> props) {
+ Long retentionMs = (Long) props.get(TopicConfig.RETENTION_MS_CONFIG);
+ Long localRetentionMs = (Long)
props.get(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG);
+ Long remoteCopyLagMs = (Long)
props.get(TopicConfig.REMOTE_COPY_LAG_MS_CONFIG);
+ long effectiveLocalRetentionMs = localRetentionMs == -2 ? retentionMs
: localRetentionMs;
+ if (remoteCopyLagMs > 0 && effectiveLocalRetentionMs >= 0
+ && remoteCopyLagMs > effectiveLocalRetentionMs) {
+ String message = String.format("Value must not exceed %s
(effective value: %d)",
+ TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG,
effectiveLocalRetentionMs);
+ throw new ConfigException(TopicConfig.REMOTE_COPY_LAG_MS_CONFIG,
remoteCopyLagMs, message);
+ }
+ }
+
+ private static void validateRemoteCopyLagSize(Map<?, ?> props) {
Review Comment:
`Map<?, ?>` ->`Map<String, ?>`
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java:
##########
@@ -608,6 +631,32 @@ private static void
validateRemoteStorageRetentionTime(Map<String, ?> props) {
}
}
+ private static void validateRemoteCopyLagTime(Map<?, ?> props) {
Review Comment:
`Map<?, ?>` ->`Map<String, ?>`
##########
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:
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?
##########
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) {
Review 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?
--
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]
