Repository: kafka Updated Branches: refs/heads/0.11.0 178f215ad -> 02c4fccf4
MINOR: Improve log warning to include the log name Author: Ismael Juma <[email protected]> Reviewers: Rajini Sivaram <[email protected]> Closes #3562 from ijuma/tweak-log-warning (cherry picked from commit 7727d7a9a574ec4305bcf9f490b329f4ffcf4824) Signed-off-by: Ismael Juma <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/02c4fccf Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/02c4fccf Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/02c4fccf Branch: refs/heads/0.11.0 Commit: 02c4fccf490cd265a8ca29a82bc4b9a12ac7dc66 Parents: 178f215 Author: Ismael Juma <[email protected]> Authored: Sat Jul 22 10:35:02 2017 +0100 Committer: Ismael Juma <[email protected]> Committed: Sat Jul 22 10:35:24 2017 +0100 ---------------------------------------------------------------------- .../src/main/scala/kafka/log/LogCleanerManager.scala | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/02c4fccf/core/src/main/scala/kafka/log/LogCleanerManager.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/log/LogCleanerManager.scala b/core/src/main/scala/kafka/log/LogCleanerManager.scala index 6e0ebfb..c200300 100755 --- a/core/src/main/scala/kafka/log/LogCleanerManager.scala +++ b/core/src/main/scala/kafka/log/LogCleanerManager.scala @@ -287,7 +287,7 @@ private[log] object LogCleanerManager extends Logging { if (offset < logStartOffset) { // don't bother with the warning if compact and delete are enabled. if (!isCompactAndDelete(log)) - warn(s"Resetting first dirty offset to log start offset $logStartOffset since the checkpointed offset $offset is invalid.") + warn(s"Resetting first dirty offset of ${log.name} to log start offset $logStartOffset since the checkpointed offset $offset is invalid.") logStartOffset } else { offset @@ -302,7 +302,7 @@ private[log] object LogCleanerManager extends Logging { // find first segment that cannot be cleaned // neither the active segment, nor segments with any messages closer to the head of the log than the minimum compaction lag time // may be cleaned - val firstUncleanableDirtyOffset: Long = Seq ( + val firstUncleanableDirtyOffset: Long = Seq( // we do not clean beyond the first unstable offset log.firstUnstableOffset.map(_.messageOffset), @@ -312,12 +312,11 @@ private[log] object LogCleanerManager extends Logging { // the first segment whose largest message timestamp is within a minimum time lag from now if (compactionLagMs > 0) { - dirtyNonActiveSegments.find { - s => - val isUncleanable = s.largestTimestamp > now - compactionLagMs - debug(s"Checking if log segment may be cleaned: log='${log.name}' segment.baseOffset=${s.baseOffset} segment.largestTimestamp=${s.largestTimestamp}; now - compactionLag=${now - compactionLagMs}; is uncleanable=$isUncleanable") - isUncleanable - } map(_.baseOffset) + dirtyNonActiveSegments.find { s => + val isUncleanable = s.largestTimestamp > now - compactionLagMs + debug(s"Checking if log segment may be cleaned: log='${log.name}' segment.baseOffset=${s.baseOffset} segment.largestTimestamp=${s.largestTimestamp}; now - compactionLag=${now - compactionLagMs}; is uncleanable=$isUncleanable") + isUncleanable + }.map(_.baseOffset) } else None ).flatten.min
