Skip to content

KAFKA-2669; Fix LogCleanerIntegrationTest #327

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
lindong28 wants to merge 6 commits into from
Closed

KAFKA-2669; Fix LogCleanerIntegrationTest #327

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
39c356a
KAFKA-2669; Fix LogCleanerIntegrationTest
Oct 17, 2015
2616cad
improve awaitCleaned as reviewers suggested
Oct 17, 2015
6cbb4a5
add timeout for awaitCleaned
Oct 18, 2015
74cd5e7
replace return with nested function
Oct 19, 2015
0f4d1cf
minor change
lindong28 Oct 19, 2015
69ffb13
address comments
lindong28 Oct 19, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions core/src/main/scala/kafka/log/LogCleaner.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,12 +149,25 @@ class LogCleaner(val config: CleanerConfig,
}

/**
* For testing, a way to know when work has completed. This method blocks until the
* For testing, a way to know when work has completed. This method waits until the
* cleaner has processed up to the given offset on the specified topic/partition
*
* @param topic The Topic to be cleaned
* @param part The partition of the topic to be cleaned
* @param offset The first dirty offset that the cleaner doesn't have to clean
* @param maxWaitMs The maximum time in ms to wait for cleaner
*
* @return A boolean indicating whether the work has completed before timeout
*/
def awaitCleaned(topic: String, part: Int, offset: Long, timeout: Long = 30000L): Unit = {
while(!cleanerManager.allCleanerCheckpoints.contains(TopicAndPartition(topic, part)))
Thread.sleep(10)
def awaitCleaned(topic: String, part: Int, offset: Long, maxWaitMs: Long = 60000L): Boolean = {

@ijuma ijuma Oct 19, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for implementing the timeout. I, personally, prefer the following implementation:

def awaitCleaned(topic: String, part: Int, offset: Long, maxWaitMs: Long = 60000L): Boolean = {
  def isCleaned = cleanerManager.allCleanerCheckpoints.get(TopicAndPartition(topic, part)).fold(false)(_ >= offset)
  var remainingWaitMs = maxWaitMs
  while (!isCleaned && remainingWaitMs > 0) {
    val sleepTime = math.min(100, remainingWaitMs)
    Thread.sleep(sleepTime)
    remainingWaitMs -= sleepTime
  }
  isCleaned

It avoids return, which is discouraged in Scala, makes the loop condition clearer and the timeout only controls when to exit the loop, the return value is based on isCleaned only.

@lindong28 lindong28 Oct 19, 2015

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you. This is a very nice usage of nested function in Scala:) I have updated the patch as you suggested.

@ijuma ijuma Oct 19, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Great, thanks. Sorry to nitpick, but you don't need the math.max in remainingWaitMs = ... since we check for > 0 (that is, it's OK for the value to be negative). Looks good otherwise.

@ijuma ijuma Oct 19, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One question: is there a reason why we are reducing the remainingWaitMs by 100 instead of the actual sleep time like in my suggested version (just curious)

@lindong28 lindong28 Oct 19, 2015

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ijuma Thanks for your suggestion:) I have replaced that statement by remainingWaitMs -= 100.

I think, w.r.t. functionality, my Thread.sleep(math.min(100, remainingWaitMs)) is exactly the same as your val sleepTime = math.min(100, remainingWaitMs) and Thread.sleep(sleepTime), right? I think the only difference is that my implementation used 1 less lines here and avoided extra variable sleepTime.

@ijuma ijuma Oct 19, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for updating it!

There are two differences:

  • You use the number 100 in two places. If someone decides that the sleep time should be a different value, they may miss the fact that they have to update it in two places.
  • You subtract 100 even if sleep time was lower because of math.min. This won't have any effect with the current loop condition, but it could change in a future update.

These minor issues add-up and make refactoring harder.

@lindong28 lindong28 Oct 19, 2015

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the first problem is minor and it is OK to change it twice. But I agree it is a valid problem. In stead of following either of our approach, I think it is better to do sleepInterval = 100 to make this an explicit configuration to solve the problem here.

I am a bit confused by your comments on the 2nd problem... I had this implementation previously that didn't have this problem:

  def awaitCleaned(topic: String, part: Int, offset: Long, maxWaitMs: Long = 60000L): Boolean = {
    def isCleaned = cleanerManager.allCleanerCheckpoints.get(TopicAndPartition(topic, part)).fold(false)(_ >= offset)
    var remainingWaitMs = maxWaitMs
    while (!isCleaned && remainingWaitMs > 0) {
      Thread.sleep(math.min(100, remainingWaitMs))
      remainingWaitMs = math.max(0, remainingWaitMs - 100)
    }
    return isCleaned
  }

Then you made the following comment:

Great, thanks. Sorry to nitpick, but you don't need the math.max in remainingWaitMs = ... since we check for > 0 (that is, it's OK for the value to be negative). Looks good otherwise.

Therefore I change the implementation to do remainingWaitMs -= 100. Actually I am fine with both solutions. But now I am confused by your latest comment that "it is not OK to have remainingWaitMs < 0". Did I misunderstand your comment?

@ijuma ijuma Oct 19, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sleepInterval = 100 is better if you want to make it possible for callers to configure that parameter. It seems to me that it's unnecessary, but up to you. It is not the worst thing in the world to have to change two things in adjacent lines, so you can also keep it as is if committers are fine with the code (there is also a slight readability penalty as users have to figure out that the 100 in both cases represent the same thing). It is a minor thing for sure, but personally I try to write code so that there are no minor things when the solution is easy. :)

I think I didn't explain myself well regarding point 2, but let's just drop it as it's subtle and it's not worth spending the time discussing it in this context.

@lindong28 lindong28 Oct 19, 2015

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ijuma Thanks:) I have updated patch to follow the code you suggested. Please have a look.

var remainingWaitMs = maxWaitMs
while (cleanerManager.allCleanerCheckpoints.get(TopicAndPartition(topic, part)).fold(true)(_ < offset)) {

@guozhangwang guozhangwang Oct 19, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What will happen if allCleanerCheckpoints does not contain the partition?

@lindong28 lindong28 Oct 19, 2015

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If allCleanerCheckpoints doesn't contain the partition, cleanerManager.allCleanerCheckpoints.get(TopicAndPartition(topic, part)).fold(true)(_ < offset) will return true and while loop continues.

@guozhangwang guozhangwang Oct 19, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was originally concerned that if the partition is not included due to some bugs then we are doomed to wait for 1 minute in practice; but then I realize this function is only going to be used in testing, so this should be fine.

if (remainingWaitMs == 0)
return false
Thread.sleep(math.min(100, remainingWaitMs))
remainingWaitMs = math.max(0, remainingWaitMs - 100)
}

@ijuma ijuma Oct 17, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The following is more idiomatic:

  while (cleanerManager.allCleanerCheckpoints.get(TopicAndPartition(topic, part)).fold(true)(_ < offset)
    Thread.sleep(10)

@lindong28 lindong28 Oct 17, 2015

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice approach. I will follow your suggestion.

return true
}

/**
Expand Down
2 changes: 1 addition & 1 deletion core/src/test/scala/unit/kafka/admin/DeleteTopicTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
17 changes: 12 additions & 5 deletions core/src/test/scala/unit/kafka/log/LogCleanerIntegrationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand All @@ -82,7 +89,6 @@ class LogCleanerIntegrationTest(compressionCodec: String) {

// we expect partition 0 to be gone
assert(!checkpoints.contains(topics(0)))

cleaner.shutdown()
}

Expand Down Expand Up @@ -111,6 +117,7 @@ class LogCleanerIntegrationTest(compressionCodec: String) {

@After
def teardown() {
time.scheduler.shutdown()
CoreUtils.rm(logDir)
}

Expand Down
1 change: 1 addition & 0 deletions core/src/test/scala/unit/kafka/utils/MockScheduler.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ class MockScheduler(val time: Time) extends Scheduler {

def shutdown() {
this synchronized {
tasks.foreach(_.fun())
tasks.clear()
}
}
Expand Down