kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jjko...@apache.org
Subject git commit: KAFKA-1327; Log cleaner metrics follow-up patch to reset dirtiest log cleanable ratio; reviewed by Jun Rao (cherry picked from commit 874620d)
Date Tue, 22 Apr 2014 00:23:21 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 89f040c8c -> 3af3efe37


KAFKA-1327; Log cleaner metrics follow-up patch to reset dirtiest log cleanable ratio; reviewed
by Jun Rao
(cherry picked from commit 874620d)


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/3af3efe3
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/3af3efe3
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/3af3efe3

Branch: refs/heads/trunk
Commit: 3af3efe3773348fd7adb8ca43f2abc5490416e55
Parents: 89f040c
Author: Joel Koshy <jjkoshy@gmail.com>
Authored: Mon Apr 21 17:06:39 2014 -0700
Committer: Joel Koshy <jjkoshy@gmail.com>
Committed: Mon Apr 21 17:12:48 2014 -0700

----------------------------------------------------------------------
 core/src/main/scala/kafka/log/LogCleaner.scala    | 10 ++++++----
 .../main/scala/kafka/log/LogCleanerManager.scala  | 18 ++++++++++--------
 2 files changed, 16 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/3af3efe3/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 b9ffe00..2faa196 100644
--- a/core/src/main/scala/kafka/log/LogCleaner.scala
+++ b/core/src/main/scala/kafka/log/LogCleaner.scala
@@ -19,7 +19,6 @@ package kafka.log
 
 import scala.collection._
 import scala.math
-import java.util.concurrent.TimeUnit
 import java.nio._
 import java.util.Date
 import java.io.File
@@ -215,6 +214,7 @@ class LogCleaner(val config: CleanerConfig,
      */
     def recordStats(id: Int, name: String, from: Long, to: Long, stats: CleanerStats) {
       this.lastStats = stats
+      cleaner.statsUnderlying.swap
       def mb(bytes: Double) = bytes / (1024*1024)
       val message = 
         "%n\tLog cleaner thread %d cleaned log %s (dirty section = [%d, %d])%n".format(id,
name, from, to) + 
@@ -260,9 +260,10 @@ private[log] class Cleaner(val id: Int,
 
   this.logIdent = "Cleaner " + id + ": "
   
-  /* stats on this cleaning */
-  val stats = new CleanerStats(time)
-  
+  /* cleaning stats - one instance for the current (or next) cleaning cycle and one for the
last completed cycle */
+  val statsUnderlying = (new CleanerStats(time), new CleanerStats(time))
+  def stats = statsUnderlying._1
+
   /* buffer used for read i/o */
   private var readBuffer = ByteBuffer.allocate(ioBufferSize)
   
@@ -304,6 +305,7 @@ private[log] class Cleaner(val id: Int,
     stats.bufferUtilization = offsetMap.utilization
     
     stats.allDone()
+
     endOffset
   }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/3af3efe3/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 514941c..e8ced6a 100644
--- a/core/src/main/scala/kafka/log/LogCleanerManager.scala
+++ b/core/src/main/scala/kafka/log/LogCleanerManager.scala
@@ -58,7 +58,7 @@ private[log] class LogCleanerManager(val logDirs: Array[File], val logs:
Pool[To
   private val pausedCleaningCond = lock.newCondition()
   
   /* a gauge for tracking the cleanable ratio of the dirtiest log */
-  private var dirtiestLogCleanableRatio = 0.0
+  @volatile private var dirtiestLogCleanableRatio = 0.0
   newGauge("max-dirty-percent", new Gauge[Int] { def value = (100 * dirtiestLogCleanableRatio).toInt
})
 
   /**
@@ -80,8 +80,7 @@ private[log] class LogCleanerManager(val logDirs: Array[File], val logs:
Pool[To
                           .map(l => LogToClean(l._1, l._2,           // create a LogToClean
instance for each
                                                lastClean.getOrElse(l._1, l._2.logSegments.head.baseOffset)))
                           .filter(l => l.totalBytes > 0)             // skip any empty
logs
-      if(!dirtyLogs.isEmpty)
-        this.dirtiestLogCleanableRatio = dirtyLogs.max.cleanableRatio
+      this.dirtiestLogCleanableRatio = if (!dirtyLogs.isEmpty) dirtyLogs.max.cleanableRatio
else 0
       val cleanableLogs = dirtyLogs.filter(l => l.cleanableRatio > l.log.config.minCleanableRatio)
// and must meet the minimum threshold for dirty byte ratio
       if(cleanableLogs.isEmpty) {
         None
@@ -126,7 +125,8 @@ private[log] class LogCleanerManager(val logDirs: Array[File], val logs:
Pool[To
             case LogCleaningInProgress =>
               inProgress.put(topicAndPartition, LogCleaningAborted)
             case s =>
-              throw new IllegalStateException(("Partiiton %s can't be aborted and pasued
since it's in %s state").format(topicAndPartition, s))
+              throw new IllegalStateException("Compaction for partition %s cannot be aborted
and paused since it is in %s state."
+                                              .format(topicAndPartition, s))
           }
       }
       while (!isCleaningInState(topicAndPartition, LogCleaningPaused))
@@ -142,17 +142,19 @@ private[log] class LogCleanerManager(val logDirs: Array[File], val logs:
Pool[To
     inLock(lock) {
       inProgress.get(topicAndPartition) match {
         case None =>
-          throw new IllegalStateException(("Partiiton %s can't be resumed since it's never
paused").format(topicAndPartition))
+          throw new IllegalStateException("Compaction for partition %s cannot be resumed
since it is not paused."
+                                          .format(topicAndPartition))
         case Some(state) =>
           state match {
             case LogCleaningPaused =>
               inProgress.remove(topicAndPartition)
             case s =>
-              throw new IllegalStateException(("Partiiton %s can't be resumed since it's
in %s state").format(topicAndPartition, s))
+              throw new IllegalStateException("Compaction for partition %s cannot be resumed
since it is in %s state."
+                                              .format(topicAndPartition, s))
           }
       }
     }
-    info("The cleaning for partition %s is resumed".format(topicAndPartition))
+    info("Compaction for partition %s is resumed".format(topicAndPartition))
   }
 
   /**
@@ -194,7 +196,7 @@ private[log] class LogCleanerManager(val logDirs: Array[File], val logs:
Pool[To
           inProgress.put(topicAndPartition, LogCleaningPaused)
           pausedCleaningCond.signalAll()
         case s =>
-          throw new IllegalStateException(("In-progress partiiton %s can't be in %s state").format(topicAndPartition,
s))
+          throw new IllegalStateException("In-progress partition %s cannot be in %s state.".format(topicAndPartition,
s))
       }
     }
   }


Mime
View raw message