mattwong949 commented on a change in pull request #10914:
URL: https://github.com/apache/kafka/pull/10914#discussion_r706478220
##########
File path: core/src/main/scala/kafka/log/LogCleanerManager.scala
##########
@@ -198,8 +199,23 @@ private[log] class LogCleanerManager(val logDirs:
Seq[File],
val cleanableLogs = dirtyLogs.filter { ltc =>
(ltc.needCompactionNow && ltc.cleanableBytes > 0) ||
ltc.cleanableRatio > ltc.log.config.minCleanableRatio
}
+
if(cleanableLogs.isEmpty) {
- None
+ val logsWithTombstonesExpired = dirtyLogs.filter {
+ case ltc =>
+ // in this case, we are probably in a low throughput situation
+ // therefore, we should take advantage of this fact and remove
tombstones if we can
+ // under the condition that the log's latest delete horizon is
less than the current time
+ // tracked
+ ltc.log.latestDeleteHorizon != RecordBatch.NO_TIMESTAMP &&
ltc.log.latestDeleteHorizon <= time.milliseconds()
Review comment:
@junrao @hachikuji @lbradstreet I've removed the logic for tracking the
latestDeleteHorizon and the deleteHorizon-triggered cleaning in
grabFilthiestCompactedLog since this part of the PR is not a part of KIP-534
--
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]
