MINOR: use monotonic clock for replica fetcher DelayedItem #8517

lbradstreet · 2020-04-19T19:56:31Z

Switches the replica fetcher to use a monotonic clock to calculate delays.

In doing so an unnecessary time conversion was removed the replica fetcher hot path. This is a small but significant (2.2%) cost when testing clusters with high replica counts when throttling has engaged (see attached profile).

We perform an unnecessary conversion in the replica fetcher hotpath.

ijuma · 2020-04-19T19:58:35Z

core/src/main/scala/kafka/utils/DelayedItem.scala

@@ -33,7 +33,11 @@ class DelayedItem(val delayMs: Long) extends Delayed with Logging {
   * The remaining delay time
   */
  def getDelay(unit: TimeUnit): Long = {
-    unit.convert(max(dueMs - Time.SYSTEM.milliseconds, 0), TimeUnit.MILLISECONDS)
+    unit.convert(getDelayMs, TimeUnit.MILLISECONDS)


Is this still used after this change?

It seems like it's only used in a test. So I suggest we remove it (as well as the Delayed implementation).

Sure thing. I'll drop it.

@ijuma I have removed getDelay and we no longer implement Delayed. In doing so I also switched the fetcher to a monotonic clock, as our existing implementation is dangerous.

ijuma · 2020-04-19T19:59:17Z

ok to test

ijuma · 2020-04-20T01:39:35Z

Same test failed in both jobs:

kafka.api.ConsumerBounceTest.testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup

Seems unrelated, but will retest just in case.

ijuma · 2020-04-20T01:39:41Z

retest this please

ijuma · 2020-04-21T00:13:33Z

Unrelated flaky test:

org.apache.kafka.streams.integration.EOSUncleanShutdownIntegrationTest.shouldWorkWithUncleanShutdownWipeOutStateStore[exactly_once_beta]

core/src/main/scala/kafka/utils/DelayedItem.scala

ijuma · 2020-04-22T05:29:06Z

core/src/main/scala/kafka/utils/DelayedItem.scala

@@ -21,24 +21,19 @@ import java.util.concurrent._

 import org.apache.kafka.common.utils.Time

-import scala.math._
+class DelayedItem(val delayMs: Long, val time: Time) extends Logging {
+  def this(delayMs: Long) = this(delayMs, Time.SYSTEM)


Do we need this? Can we not pass the time from every caller?

We can, I was just trying to avoid the replica fetcher having to pass Time.SYSTEM in everywhere. If you prefer that I can change it. Is there any downside what I did though?

It should pass the actual time instance so that we can mock if we want. It has one in its constructor already:

class ReplicaFetcherThread(name: String, fetcherId: Int, sourceBroker: BrokerEndPoint, brokerConfig: KafkaConfig, failedPartitions: FailedPartitions, replicaMgr: ReplicaManager, metrics: Metrics, time: Time,

Thanks, yes, I was just coming back to say your suggestion was better for testability reasons.

ijuma · 2020-04-22T13:16:24Z

ok to test

ijuma · 2020-04-22T15:20:49Z

core/src/main/scala/kafka/utils/DelayedItem.scala

@@ -21,24 +21,19 @@ import java.util.concurrent._

 import org.apache.kafka.common.utils.Time

-import scala.math._
+class DelayedItem(val delayMs: Long, val time: Time) extends Logging {


Does time need to be a public val?

Good point. I'll make it private.

ijuma · 2020-04-23T06:23:01Z

core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala

@@ -60,6 +61,7 @@ class ReplicaAlterLogDirsThreadTest {

    when(replicaManager.futureLogExists(t1p0)).thenReturn(false)

+    val time = new SystemTime()


We should use Time.SYSTEM, not create a new instance.

We should consider making SystemTime package private.

Makes sense, I followed the abstract fetcher tests which created new SystemTime(s) too. I'll see if it's easy to make SystemTime package private.

lbradstreet · 2020-04-23T07:04:56Z

I updated the PR to pass Time everywhere that's required. Note that I did not switch the fetcher tests to use MockTime as it appears they are timing dependent and may depend on AbstractFetcherThread's use of await, so I would want to make the change carefully.

ijuma

LGTM, thanks!

ijuma · 2020-04-23T13:58:41Z

retest this please

lbradstreet · 2020-05-13T19:16:45Z

@ijuma can this be merged?

ijuma · 2020-05-13T19:32:57Z

ok to test

ijuma · 2020-05-13T21:18:15Z

@lbradstreet Compile errors:

12:38:04 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/tools/src/main/java/org/apache/kafka/tools/PushHttpMetricsReporter.java:29: error: SystemTime is not public in org.apache.kafka.common.utils; cannot be accessed from outside package
12:38:04 import org.apache.kafka.common.utils.SystemTime;
12:38:04 ^
12:38:04 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/tools/src/main/java/org/apache/kafka/trogdor/workload/SustainedConnectionWorker.java:33: error: SystemTime is not public in org.apache.kafka.common.utils; cannot be accessed from outside package
12:38:04 import org.apache.kafka.common.utils.SystemTime;
12:38:04 ^
12:38:04 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/tools/src/main/java/org/apache/kafka/trogdor/workload/SustainedConnectionWorker.java:63: error: SystemTime is not public in org.apache.kafka.common.utils; cannot be accessed from outside package
12:38:04 private static final SystemTime SYSTEM_TIME = new SystemTime();
12:38:04 ^
12:38:04 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/tools/src/main/java/org/apache/kafka/tools/PushHttpMetricsReporter.java:98: error: SystemTime is not public in org.apache.kafka.common.utils; cannot be accessed from outside package
12:38:04 time = new SystemTime();
12:38:04 ^
12:38:04 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/tools/src/main/java/org/apache/kafka/trogdor/workload/SustainedConnectionWorker.java:63: error: SystemTime is not public in org.apache.kafka.common.utils; cannot be accessed from outside package
12:38:04 private static final SystemTime SYSTEM_TIME = new SystemTime();

lbradstreet · 2020-05-13T21:33:50Z

@ijuma thanks, I probably should have merged in before asking.

lbradstreet · 2020-05-13T23:15:06Z

@ijuma the updated PR should compile OK. I removed the change to make SystemTime package private as it seems to be used in other tooling now.

ijuma · 2020-05-13T23:38:56Z

ok to test

ijuma · 2020-05-15T22:33:03Z

retest this please

ijuma · 2020-05-17T02:01:36Z

ok to test

ijuma · 2020-05-17T05:30:11Z

retest this please

chia7712 · 2020-05-18T09:36:21Z

EmbeddedKafkaCluster created by Stream UT uses MockTime to create KafkaServer(
https://github.com/apache/kafka/blob/trunk/streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java#L81) so the check of delayed item gets impacted by MockTime.

lbradstreet · 2020-05-18T15:13:39Z

EmbeddedKafkaCluster created by Stream UT uses MockTime to create KafkaServer(
https://github.com/apache/kafka/blob/trunk/streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java#L81) so the check of delayed item gets impacted by MockTime.

@chia7712 thanks, nice catch. That does make things trickier.

MINOR: avoid unnecessary delay conversion in isDelayed check

f8ca76c

We perform an unnecessary conversion in the replica fetcher hotpath.

ijuma reviewed Apr 19, 2020

View reviewed changes

Use monotonic clock for delay checks

d3d9395

lbradstreet changed the title ~~MINOR: avoid unnecessary delay conversion in isDelayed check~~ MINOR: use monotonic clock for replica fetcher DelayedItem Apr 21, 2020

chia7712 reviewed Apr 21, 2020

View reviewed changes

core/src/main/scala/kafka/utils/DelayedItem.scala Outdated Show resolved Hide resolved

core/src/main/scala/kafka/utils/DelayedItem.scala Outdated Show resolved Hide resolved

chia7712 reviewed Apr 21, 2020

View reviewed changes

core/src/main/scala/kafka/utils/DelayedItem.scala Outdated Show resolved Hide resolved

chia7712 reviewed Apr 21, 2020

View reviewed changes

core/src/main/scala/kafka/utils/DelayedItem.scala Show resolved Hide resolved

Simplify delay check, fix toString delayMs, compareTo DelayedItem

ae959a9

chia7712 reviewed Apr 21, 2020

View reviewed changes

core/src/main/scala/kafka/utils/DelayedItem.scala Outdated Show resolved Hide resolved

chia7712 approved these changes Apr 21, 2020

View reviewed changes

Drop unneeded compareTo

1f9bec8

ijuma reviewed Apr 22, 2020

View reviewed changes

Supply Time to DelayedItem. Allow ReplicaAlterLogDirsThread to take Time

c09ad3d

ijuma reviewed Apr 23, 2020

View reviewed changes

lbradstreet added 2 commits April 23, 2020 06:09

Supply Time to DelayedItem. Allow ReplicaAlterLogDirsThread to take Time

e3fe2ff

Switch from SystemTime to Time.SYSTEM. Make SystemTime package private

acbc994

ijuma approved these changes Apr 23, 2020

View reviewed changes

Merge branch 'trunk' into avoid-delay-conversion

23d5726

Return SystemTime to public as it's used elsewhere now

f469898

andrewchoi5 approved these changes Jul 6, 2020

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

MINOR: use monotonic clock for replica fetcher DelayedItem #8517

MINOR: use monotonic clock for replica fetcher DelayedItem #8517

lbradstreet commented Apr 19, 2020 •

edited

ijuma Apr 19, 2020

ijuma Apr 21, 2020

lbradstreet Apr 21, 2020

lbradstreet Apr 21, 2020

ijuma commented Apr 19, 2020

ijuma commented Apr 20, 2020

ijuma commented Apr 20, 2020

ijuma commented Apr 21, 2020

ijuma Apr 22, 2020

lbradstreet Apr 22, 2020

ijuma Apr 22, 2020 •

edited

lbradstreet Apr 22, 2020

ijuma commented Apr 22, 2020

ijuma Apr 22, 2020 •

edited

lbradstreet Apr 22, 2020

ijuma Apr 23, 2020

ijuma Apr 23, 2020

lbradstreet Apr 23, 2020 •

edited

lbradstreet commented Apr 23, 2020 •

edited

ijuma left a comment

ijuma commented Apr 23, 2020

lbradstreet commented May 13, 2020

ijuma commented May 13, 2020

ijuma commented May 13, 2020

lbradstreet commented May 13, 2020

lbradstreet commented May 13, 2020

ijuma commented May 13, 2020

ijuma commented May 15, 2020

ijuma commented May 17, 2020

ijuma commented May 17, 2020

chia7712 commented May 18, 2020

lbradstreet commented May 18, 2020

		@@ -60,6 +61,7 @@ class ReplicaAlterLogDirsThreadTest {

		when(replicaManager.futureLogExists(t1p0)).thenReturn(false)

		val time = new SystemTime()

MINOR: use monotonic clock for replica fetcher DelayedItem #8517

Are you sure you want to change the base?

MINOR: use monotonic clock for replica fetcher DelayedItem #8517

Conversation

lbradstreet commented Apr 19, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma commented Apr 19, 2020

ijuma commented Apr 20, 2020

ijuma commented Apr 20, 2020

ijuma commented Apr 21, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma Apr 22, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma commented Apr 22, 2020

ijuma Apr 22, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lbradstreet Apr 23, 2020 • edited

Choose a reason for hiding this comment

lbradstreet commented Apr 23, 2020 • edited

ijuma left a comment

Choose a reason for hiding this comment

ijuma commented Apr 23, 2020

lbradstreet commented May 13, 2020

ijuma commented May 13, 2020

ijuma commented May 13, 2020

lbradstreet commented May 13, 2020

lbradstreet commented May 13, 2020

ijuma commented May 13, 2020

ijuma commented May 15, 2020

ijuma commented May 17, 2020

ijuma commented May 17, 2020

chia7712 commented May 18, 2020

lbradstreet commented May 18, 2020

lbradstreet commented Apr 19, 2020 •

edited

ijuma Apr 22, 2020 •

edited

ijuma Apr 22, 2020 •

edited

lbradstreet Apr 23, 2020 •

edited

lbradstreet commented Apr 23, 2020 •

edited