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 2 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
4 changes: 2 additions & 2 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 @@ -152,8 +152,8 @@ class LogCleaner(val config: CleanerConfig,
* For testing, a way to know when work has completed. This method blocks until the
* 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)))
def awaitCleaned(topic: String, part: Int, offset: Long): Unit = {

@auradkar auradkar Oct 17, 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.

Can you check if this is only used in a test? Ideally, we should use the timeout to prevent indefinite blocking

@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.

Yeah, we don't want the test to hang forever instead of failing with a time out.

@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.

Sure. I have verified that this is only used in test.

@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.

Also note that the "timeout" is not specified in API, not used in any invocation, or implemented in the function. If I were to implement this, I need to find a good default value that doesn't break existing test. Since it is only used in the test, I prefer to leave it as it is and only implement it when we have a good usecase.

@ijuma ijuma Oct 18, 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.

@lindong28 The bug here is that the timeout was not implemented in the function. Because there is a default value (30000L), callers are expecting that timeout so I don't agree with your assessment that it is not used in invocations. This pattern is really common in our tests (see TestUtils.waitUntil* methods) and callers usually don't override the default timeout. Given that, I think we already have a use-case, it is why the parameter exists in the first place.

@lindong28 lindong28 Oct 18, 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 Yeah I know we have default value of 30000L for timeout. My point is that, since it is not explicitly specified in the invocation it probably means users don't care about it. It is important that, if we enable this timeout, existing tests won't fail due to low timeout value.

I am not sure.. but can you explain why 30000 ms is a good default value for timeout? Note that log cleaner's default backoffMs is 15000 ms.

Alternatively, how about I set timeout to be LONG.MAX by default? This won't accidentally fail existing test while still allow developers to use timeout when they want.

@ijuma ijuma Oct 18, 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.

I disagree with the assertion that because the callers didn't explicitly set the timeout, they don't care about a timeout. We have several examples of callers of TestUtils.waitUntilTrue that care about a timeout, but don't set it explicitly because the default is OK (that's a big reason why default arguments are useful).

I don't know what is a good default for this method, but I certainly know that we can't wait indefinitely for things in tests. It means that a bug can cause the whole test suite to hang, which is a very bad outcome (even worse than accidentally failing tests, which is also bad).

Ideally we'd set a timeout that would not cause tests to fail, but that wouldn't delay the test suite by too long if it were to fail. It's OK to err on the side of caution, but Long.MaxValue is far too long. Why not choose a value that you think is appropriate and then run the test suite a few times in a loop to see if it's OK?

@lindong28 lindong28 Oct 18, 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 have already run LogCleanerIntegrationTest and DeleteTopicTest a few times with infinite timeout and the tests run well. That is the reason why I think inifinite timeout works for existing tests. I don't think setting a default timeout value that is large enough for existing invocation in tests is a very good idea -- a default time value should ideally be good for most usecase from first principle.

But I agree with your point that we should never allow tests to run forever. How about I make the following change:

  1. give default timeout value of 60*1000 ms. This is 4X the default backoffMs which is long enough for existing tests.

  2. let awaitCleaned return a boolean value to indicate whether it has timed out or not -- this allows the tests to fail in case of timeout so that we can look into the problem.

@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.

Chiming late here: I also feel setting an infinite timeout value is generally not a good idea since IF there are any issue causing the test to block, it is hard to detect this issue: for example today we also encounter blocking-forever test cases in Jenkins from time to time, and because we set the Jenkins time to 1 hour (? not sure if the value is exact), we ended up seeing an "timed out" failed Jenkins without much information which test case caused it.

I think the current approach of setting a large enough default value is a better option.

@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.

Thanks for the comment. I definitely agree that we should prevent test from running forever.

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

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