Repository: kafka Updated Branches: refs/heads/trunk f4ac68a40 -> 1f6494202
KAFKA-6030; Fix Integer overflow in cleanable ratio computation Author: Xin Li <xin...@trivago.com> Reviewers: Ismael Juma <ism...@juma.me.uk>, Jason Gustafson <ja...@confluent.io> Closes #4043 from lisa2lisa/fix (cherry picked from commit bb27215ceac9caaac446b5fe43d3f3284624fcb4) Signed-off-by: Jason Gustafson <ja...@confluent.io> Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/1f649420 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/1f649420 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/1f649420 Branch: refs/heads/trunk Commit: 1f64942021bd8fe0f2791700fb402cfac1acc9de Parents: f4ac68a Author: Xin Li <xin...@trivago.com> Authored: Mon Oct 9 14:15:44 2017 -0700 Committer: Jason Gustafson <ja...@confluent.io> Committed: Mon Oct 9 14:33:24 2017 -0700 ---------------------------------------------------------------------- core/src/main/scala/kafka/log/LogCleaner.scala | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/1f649420/core/src/main/scala/kafka/log/LogCleaner.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/log/LogCleaner.scala b/core/src/main/scala/kafka/log/LogCleaner.scala index 217c49e..3ab244b 100644 --- a/core/src/main/scala/kafka/log/LogCleaner.scala +++ b/core/src/main/scala/kafka/log/LogCleaner.scala @@ -650,9 +650,9 @@ private[log] class Cleaner(val id: Int, var segs = segments.toList while(segs.nonEmpty) { var group = List(segs.head) - var logSize = segs.head.size - var indexSize = segs.head.index.sizeInBytes - var timeIndexSize = segs.head.timeIndex.sizeInBytes + var logSize = segs.head.size.toLong + var indexSize = segs.head.index.sizeInBytes.toLong + var timeIndexSize = segs.head.timeIndex.sizeInBytes.toLong segs = segs.tail while(segs.nonEmpty && logSize + segs.head.size <= maxSize && @@ -847,10 +847,10 @@ private class CleanerStats(time: Time = Time.SYSTEM) { * Helper class for a log, its topic/partition, the first cleanable position, and the first uncleanable dirty position */ private case class LogToClean(topicPartition: TopicPartition, log: Log, firstDirtyOffset: Long, uncleanableOffset: Long) extends Ordered[LogToClean] { - val cleanBytes = log.logSegments(-1, firstDirtyOffset).map(_.size).sum + val cleanBytes = log.logSegments(-1, firstDirtyOffset).map(_.size.toLong).sum private[this] val firstUncleanableSegment = log.logSegments(uncleanableOffset, log.activeSegment.baseOffset).headOption.getOrElse(log.activeSegment) val firstUncleanableOffset = firstUncleanableSegment.baseOffset - val cleanableBytes = log.logSegments(firstDirtyOffset, math.max(firstDirtyOffset, firstUncleanableOffset)).map(_.size).sum + val cleanableBytes = log.logSegments(firstDirtyOffset, math.max(firstDirtyOffset, firstUncleanableOffset)).map(_.size.toLong).sum val totalBytes = cleanBytes + cleanableBytes val cleanableRatio = cleanableBytes / totalBytes.toDouble override def compare(that: LogToClean): Int = math.signum(this.cleanableRatio - that.cleanableRatio).toInt
