KAFKA-10040; Make computing the PreferredReplicaImbalanceCount metric more efficient by dajac · Pull Request #8724 · apache/kafka

dajac · 2020-05-25T15:26:33Z

This PR changes the way PreferredReplicaImbalanceCount is computed. It moves from re-computing after the processing of each event in the controller to continuously maintaining the count.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

… more efficient

hachikuji · 2020-05-26T16:29:48Z

ok to test

stanislavkozlovski · 2020-05-31T21:32:28Z

+  private def updatePreferredReplicaImbalanceMetric(partition: TopicPartition,
+                                                    oldReplicaAssignment: Option[ReplicaAssignment],
+                                                    oldLeadershipInfo: Option[LeaderIsrAndControllerEpoch],
+                                                    newReplicaAssignment: Option[ReplicaAssignment],


nit: I think this exact argument doesn't need to be an Option

Actually, in updatePartitionFullReplicaAssignment, partitionLeadershipInfo.get(topicPartition) returns an Option so it is more convenient to keep it as an Option. Ok for you?

hachikuji · 2020-06-03T01:26:03Z

+    }
+  }
+
+  private def isReplicaImbalance(replicaAssignment: ReplicaAssignment,


What about hasPreferredLeader?

Yes, that is fine.

hachikuji · 2020-06-03T01:34:56Z

+                                 leadershipInfo: LeaderIsrAndControllerEpoch): Boolean = {
+    val preferredReplica = replicaAssignment.replicas.head
+    if (replicaAssignment.isBeingReassigned && replicaAssignment.addingReplicas.contains(preferredReplica))
+      // reassigning partitions are not counted as imbalanced until the new replica joins the ISR (completes reassignment)


Not something to change here, but it's a little curious that we only do this when a reassignment is in progress. I'm not sure it's useful to distinguish that case as opposed to a broker that is catching up after a restart. If the preferred leader is not in the ISR, then we can't elect it anyway. Seems like it might be more useful if this metric captured the "eligible imbalance".

I agree with you.

hachikuji · 2020-06-03T01:38:18Z

    partitionsForTopic(topic).filter { partition => states.contains(partitionState(partition)) }.toSet
  }

+  def putLeadershipInfo(partition: TopicPartition, leaderIsrAndControllerEpoch: LeaderIsrAndControllerEpoch): Unit = {


Wonder if we should consider making partitionLeadershipInfo private so that we do not mistakenly invoke put in the future directly.

That makes total sense. I have updated the PR to make it private. I have added methods in the controller context for all the reads (moved code around).

hachikuji · 2020-06-03T01:40:10Z

      if (!isActive) {
        0
      } else {
-        controllerContext.allPartitions.count { topicPartition =>


Ouch. So the main issue is that we were doing this full iteration over all partitions on every controller event.

dajac · 2020-06-03T07:27:32Z

@hachikuji @stanislavkozlovski I have addressed your comments. Could you have a second look?

hachikuji · 2020-06-03T16:22:13Z

ok to test

hachikuji · 2020-06-03T16:23:18Z

retest this please

hachikuji · 2020-06-03T16:23:27Z

retest this please

hachikuji

Thanks for the updates. Just a couple small comments.

hachikuji · 2020-06-03T16:32:48Z

+    }.keySet
+  }
+
+  def partitionLeadsOnBroker(brokerId: Int): Set[TopicPartition] = {


nit: partitionLeadersOnBroker?

That makes sense.

hachikuji · 2020-06-03T16:46:02Z

+  }
+
+  def partitionsWithLeaders(): Set[TopicPartition] = {
+    partitionLeadershipInfo.keySet


This definition seems inconsistent with partitionsWithoutLeaders. I think you're just trying to preserve the existing logic. It might make sense to use a different name to avoid the apparent inconsistency? Maybe we could change partitionsWithoutLeaders to be partitionsWithOfflineLeaders or something like that. Looking at the caller, it looks like it would be reasonable to exclude topics which are being queued for deletion in both cases, but we could change that separately if you think it is risky.

Right. I did not pay attention to this detail but that makes total sense to use a better name here. Let's go with partitionsWithOfflineLeaders.
We can indeed exclude topics which are being queued for deletion. It is already done in partitionsWithOfflineLeaders but I have added it in partitionsWithLeaders.

dajac · 2020-06-04T08:50:25Z

@hachikuji Thanks for the review. I have updated the PR based on your feedback.

hachikuji · 2020-06-04T16:23:35Z

ok to test

hachikuji

LGTM

hachikuji · 2020-06-04T18:31:02Z

retest this please

hachikuji · 2020-06-05T00:29:26Z

I will go ahead and merge this. I build locally with JDK11 and Scala 2.13.

* 'trunk' of github.com:apache/kafka: (46 commits) MINOR: improve code encapsulation between StreamThread and TaskManager (apache#8819) Fixing KAFKA-10094 (apache#8797) KAFKA-9851: Revoking Connect tasks due to connectivity issues should also clear the running assignment (apache#8804) KAFKA-9840; Skip End Offset validation when the leader epoch is not reliable (apache#8486) HOT_FIX: Update javadoc since imports added (apache#8817) KAFKA-8011: Fix flaky RegexSourceIntegrationTest (apache#8799) KAFKA-9570: Define SSL configs in all worker config classes, not just distributed (apache#8135) KAFKA-10111: Make SinkTaskContext.errantRecordReporter() a default method (apache#8814) KAFKA-10110: Corrected potential NPE when null label value added to KafkaMetricsContext (apache#8811) MINOR: Change the order that Connect calls `config()` and `validate()` to avoid validating if the required ConfigDef is null (apache#8810) MINOR: fix backwards incompatibility in JmxReporter introduced by KIP-606 MINOR: Fix javadoc warnings (apache#8809) KAFKA-9441: Improve Kafka Streams task management (apache#8776) fix the broken links of streams javadoc (apache#8789) KAFKA-10040; Make computing the PreferredReplicaImbalanceCount metric more efficient (apache#8724) KAFKA-10066: TestOutputTopic should pass record headers into deserializers (apache#8759) MINOR: Add explanation for disabling forwarding from value transformers (apache#8771) KAFKA-10033: Throw UnknownTopicOrPartitionException if altering configs of non-existing topic KAFKA-9434: automated protocol for alterReplicaLogDirs (apache#8311) KAFKA-9313: Set `use_all_dns_ips` as the new default for `client.dns.lookup` (KIP-602) (apache#8644) ...

Conflicts: * build.gradle: take upstream changes regarding heap memory configuration for the build. * apache-github/trunk: (33 commits) MINOR: fix HTML markup (apache#8823) KAFKA-10012; Reduce overhead of strings in SelectorMetrics (apache#8684) KAFKA-9216: Enforce internal config topic settings for Connect workers during startup (apache#8270) KAFKA-10097: Internalize checkpoint data (apache#8820) KAFKA-10033: Throw UnknownTopicOrPartitionException when modifying a non-existent topic's config MINOR: improve code encapsulation between StreamThread and TaskManager (apache#8819) Fixing KAFKA-10094 (apache#8797) KAFKA-9851: Revoking Connect tasks due to connectivity issues should also clear the running assignment (apache#8804) KAFKA-9840; Skip End Offset validation when the leader epoch is not reliable (apache#8486) HOT_FIX: Update javadoc since imports added (apache#8817) KAFKA-8011: Fix flaky RegexSourceIntegrationTest (apache#8799) KAFKA-9570: Define SSL configs in all worker config classes, not just distributed (apache#8135) KAFKA-10111: Make SinkTaskContext.errantRecordReporter() a default method (apache#8814) KAFKA-10110: Corrected potential NPE when null label value added to KafkaMetricsContext (apache#8811) MINOR: Change the order that Connect calls `config()` and `validate()` to avoid validating if the required ConfigDef is null (apache#8810) MINOR: fix backwards incompatibility in JmxReporter introduced by KIP-606 MINOR: Fix javadoc warnings (apache#8809) KAFKA-9441: Improve Kafka Streams task management (apache#8776) fix the broken links of streams javadoc (apache#8789) KAFKA-10040; Make computing the PreferredReplicaImbalanceCount metric more efficient (apache#8724) ...

dajac added 2 commits May 25, 2020 17:10

KAFKA-10040; Make computing the PreferredReplicaImbalanceCount metric…

c2fb87a

… more efficient

refactoring

f2c9df9

ijuma requested a review from junrao May 28, 2020 04:04

stanislavkozlovski reviewed May 31, 2020

View reviewed changes

hachikuji reviewed Jun 3, 2020

View reviewed changes

refactor

303bb10

cleanuo

8c86178

hachikuji reviewed Jun 3, 2020

View reviewed changes

address review

4f01642

hachikuji approved these changes Jun 4, 2020

View reviewed changes

hachikuji merged commit 7867c24 into apache:trunk Jun 5, 2020

dajac deleted the KAFKA-10040 branch June 5, 2020 05:32

Uh oh!

Conversation

dajac commented May 25, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

hachikuji commented May 26, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji Jun 3, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dajac commented Jun 3, 2020

Uh oh!

hachikuji commented Jun 3, 2020

Uh oh!

hachikuji commented Jun 3, 2020

Uh oh!

hachikuji commented Jun 3, 2020

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dajac commented Jun 4, 2020

Uh oh!

hachikuji commented Jun 4, 2020

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

hachikuji commented Jun 4, 2020

Uh oh!

hachikuji commented Jun 5, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

dajac commented May 25, 2020 •

edited

Loading

hachikuji Jun 3, 2020 •

edited

Loading