KAFKA-2669; Fix LogCleanerIntegrationTest

core/src/main/scala/kafka/log/LogCleaner.scala

@@ -153,8 +153,11 @@ class LogCleaner(val config: CleanerConfig,
    * cleaner has processed up to the given offset on the specified topic/partition
    */
   def awaitCleaned(topic: String, part: Int, offset: Long, timeout: Long = 30000L): Unit = {
-    while(!cleanerManager.allCleanerCheckpoints.contains(TopicAndPartition(topic, part)))
-      Thread.sleep(10)
+    while(true) {
+      val cleanedOffset = cleanerManager.allCleanerCheckpoints.get(TopicAndPartition(topic, part))
+      if (cleanedOffset == None || cleanedOffset.get < offset) Thread.sleep(10)
+      else return
+    }
   }
 
   /**

core/src/test/scala/unit/kafka/admin/DeleteTopicTest.scala

@@ -239,7 +239,7 @@ class DeleteTopicTest extends ZooKeeperTestHarness {
     writeDups(numKeys = 100, numDups = 3,log)
 
     // wait for cleaner to clean
-   server.logManager.cleaner.awaitCleaned(topicName,0,0)
+   server.logManager.cleaner.awaitCleaned(topicName, 0, 0)
 
     // delete topic
     AdminUtils.deleteTopic(zkClient, "test")

core/src/test/scala/unit/kafka/log/LogCleanerIntegrationTest.scala

@@ -56,18 +56,25 @@ class LogCleanerIntegrationTest(compressionCodec: String) {
     val startSize = log.size
     cleaner.startup()
 
-    val lastCleaned = log.activeSegment.baseOffset
+    val firstDirty = log.activeSegment.baseOffset
     // wait until we clean up to base_offset of active segment - minDirtyMessages
-    cleaner.awaitCleaned("log", 0, lastCleaned)
+    cleaner.awaitCleaned("log", 0, firstDirty)
+
+    val lastCleaned = cleaner.cleanerManager.allCleanerCheckpoints.get(TopicAndPartition("log", 0)).get
+    assertTrue("log cleaner should have processed up to offset " + firstDirty, lastCleaned >= firstDirty);
 
     val read = readFromLog(log)
     assertEquals("Contents of the map shouldn't change.", appends.toMap, read.toMap)
     assertTrue(startSize > log.size)
 
     // write some more stuff and validate again
     val appends2 = appends ++ writeDups(numKeys = 100, numDups = 3, log, CompressionCodec.getCompressionCodec(compressionCodec))
-    val lastCleaned2 = log.activeSegment.baseOffset
-    cleaner.awaitCleaned("log", 0, lastCleaned2)
+    val firstDirty2 = log.activeSegment.baseOffset
+    cleaner.awaitCleaned("log", 0, firstDirty2)
+
+    val lastCleaned2 = cleaner.cleanerManager.allCleanerCheckpoints.get(TopicAndPartition("log", 0)).get
+    assertTrue("log cleaner should have processed up to offset " + firstDirty2, lastCleaned2 >= firstDirty2);
+
     val read2 = readFromLog(log)
     assertEquals("Contents of the map shouldn't change.", appends2.toMap, read2.toMap)
 
@@ -82,7 +89,6 @@ class LogCleanerIntegrationTest(compressionCodec: String) {
 
     // we expect partition 0 to be gone
     assert(!checkpoints.contains(topics(0)))
-
     cleaner.shutdown()
   }
 
@@ -111,6 +117,7 @@ class LogCleanerIntegrationTest(compressionCodec: String) {
 
   @After
   def teardown() {
+    time.scheduler.shutdown()
     CoreUtils.rm(logDir)
   }
 

core/src/test/scala/unit/kafka/utils/MockScheduler.scala

@@ -43,6 +43,7 @@ class MockScheduler(val time: Time) extends Scheduler {
 
   def shutdown() {
     this synchronized {
+      tasks.foreach(_.fun())
       tasks.clear()
     }
   }