KAFKA-4051: Use nanosecond clock for timers in broker #1768
Conversation
|
This is a small change to the broker to fix the specific issue reported in the JIRA where broker restarts are necessary to handle consumer rebalance when the clock is moved backwards by a significant amount. Note that this PR does not address the dependency on wall-clock time in Java clients. I have run manual tests on Linux and Mac to ensure that the issue reported in the JIRA is fixed by this PR. Also ran benchmark tests of the system test runs to compare the results with and without the PR, and comparing the results over a couple of runs, there is no noticeable impact beyond the differences between test runs. The results with the PR are https://jenkins.confluent.io/job/system-test-kafka-branch-builder/501/ and the results of the same trunk level without the PR are https://jenkins.confluent.io/job/system-test-kafka-branch-builder/500/. |
|
LGTM. Thank you! |
| @@ -44,6 +46,8 @@ trait Time { | |||
|
|
|||
| def nanoseconds: Long | |||
|
|
|||
| def hiResClockMs: Long = TimeUnit.NANOSECONDS.toMillis(nanoseconds) | |||
There was a problem hiding this comment.
Why have we implemented this here instead of SystemTime with a different implementation for MockTime?
There was a problem hiding this comment.
It seems that for the Scala implementation, the mock time implementation should just return the same time unless sleep is called. For the Java one, we have an optional autoTick functionality, but we don't care about that here.
There was a problem hiding this comment.
@ijuma Thank you for looking into this. Since MockTime.scala maintains only one time, shall I leave the common implementation above in Time.scala?
There was a problem hiding this comment.
OK, I understand now, my bad. You are relying on Time.nanoseconds in your implementation instead of System.nanoTime. Thanks for explaining. :)
Use System.nanoseconds instead of System.currentTimeMillis in broker timer tasks to cope with changes to wall-clock time.