kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ij...@apache.org
Subject kafka git commit: KAFKA-6146; minimize the number of triggers enqueuing PreferredReplicaLeaderElection events
Date Thu, 09 Nov 2017 00:57:20 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk ee1aaa091 -> fef80c863


KAFKA-6146; minimize the number of triggers enqueuing PreferredReplicaLeaderElection events

We currently enqueue a PreferredReplicaLeaderElection controller event in PreferredReplicaElectionHandler's
handleCreation, handleDeletion, and handleDataChange. We can just enqueue the event upon znode
creation and after preferred replica leader election completes. The processing of this latter
enqueue will register the exist watch on PreferredReplicaElectionZNode and perform any pending
preferred replica leader election that may have occurred between completion and registration.

Author: Onur Karaman <okaraman@linkedin.com>

Reviewers: Jun Rao <junrao@gmail.com>, Ismael Juma <ismael@juma.me.uk>

Closes #4189 from onurkaraman/KAFKA-6146


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/fef80c86
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/fef80c86
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/fef80c86

Branch: refs/heads/trunk
Commit: fef80c8636d03a25a3b17f31d9cf2ab9b98f385f
Parents: ee1aaa0
Author: Onur Karaman <okaraman@linkedin.com>
Authored: Thu Nov 9 00:57:15 2017 +0000
Committer: Ismael Juma <ismael@juma.me.uk>
Committed: Thu Nov 9 00:57:15 2017 +0000

----------------------------------------------------------------------
 .../kafka/controller/KafkaController.scala      | 25 +++++++-----
 .../controller/ControllerIntegrationTest.scala  | 43 ++++++++++++++------
 2 files changed, 45 insertions(+), 23 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/fef80c86/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala
index aa5cf1f..845ac63 100644
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -837,8 +837,11 @@ class KafkaController(val config: KafkaConfig, zkClient: KafkaZkClient,
time: Ti
         warn("Partition %s failed to complete preferred replica leader election. Leader is
%d".format(partition, currentLeader))
       }
     }
-    if (!isTriggeredByAutoRebalance)
+    if (!isTriggeredByAutoRebalance) {
       zkClient.deletePreferredReplicaElection()
+      // Ensure we detect future preferred replica leader elections
+      eventManager.put(PreferredReplicaLeaderElection)
+    }
   }
 
   /**
@@ -1361,14 +1364,18 @@ class KafkaController(val config: KafkaConfig, zkClient: KafkaZkClient,
time: Ti
 
     override def process(): Unit = {
       if (!isActive) return
-      zkClient.registerZNodeChangeHandlerAndCheckExistence(preferredReplicaElectionHandler)
-      val partitions = zkClient.getPreferredReplicaElection
-      val partitionsForTopicsToBeDeleted = partitions.filter(p => topicDeletionManager.isTopicQueuedUpForDeletion(p.topic))
-      if (partitionsForTopicsToBeDeleted.nonEmpty) {
-        error("Skipping preferred replica election for partitions %s since the respective
topics are being deleted"
-          .format(partitionsForTopicsToBeDeleted))
+
+      // We need to register the watcher if the path doesn't exist in order to detect future
preferred replica
+      // leader elections and we get the `path exists` check for free
+      if (zkClient.registerZNodeChangeHandlerAndCheckExistence(preferredReplicaElectionHandler))
{
+        val partitions = zkClient.getPreferredReplicaElection
+        val partitionsForTopicsToBeDeleted = partitions.filter(p => topicDeletionManager.isTopicQueuedUpForDeletion(p.topic))
+        if (partitionsForTopicsToBeDeleted.nonEmpty) {
+          error("Skipping preferred replica election for partitions %s since the respective
topics are being deleted"
+            .format(partitionsForTopicsToBeDeleted))
+        }
+        onPreferredReplicaElection(partitions -- partitionsForTopicsToBeDeleted)
       }
-      onPreferredReplicaElection(partitions -- partitionsForTopicsToBeDeleted)
     }
   }
 
@@ -1480,8 +1487,6 @@ class PreferredReplicaElectionHandler(controller: KafkaController, eventManager:
   override val path: String = PreferredReplicaElectionZNode.path
 
   override def handleCreation(): Unit = eventManager.put(controller.PreferredReplicaLeaderElection)
-  override def handleDeletion(): Unit = eventManager.put(controller.PreferredReplicaLeaderElection)
-  override def handleDataChange(): Unit = eventManager.put(controller.PreferredReplicaLeaderElection)
 }
 
 class ControllerChangeHandler(controller: KafkaController, eventManager: ControllerEventManager)
extends ZNodeChangeHandler {

http://git-wip-us.apache.org/repos/asf/kafka/blob/fef80c86/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala b/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala
index 99cacb3..d917fe1 100644
--- a/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala
+++ b/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala
@@ -200,21 +200,23 @@ class ControllerIntegrationTest extends ZooKeeperTestHarness {
   def testPreferredReplicaLeaderElection(): Unit = {
     servers = makeServers(2)
     val controllerId = TestUtils.waitUntilControllerElected(zkUtils)
-    val otherBrokerId = servers.map(_.config.brokerId).filter(_ != controllerId).head
+    val otherBroker = servers.find(_.config.brokerId != controllerId).get
     val tp = TopicAndPartition("t", 0)
-    val assignment = Map(tp.partition -> Seq(otherBrokerId, controllerId))
+    val assignment = Map(tp.partition -> Seq(otherBroker.config.brokerId, controllerId))
     TestUtils.createTopic(zkUtils, tp.topic, partitionReplicaAssignment = assignment, servers
= servers)
-    servers(otherBrokerId).shutdown()
-    servers(otherBrokerId).awaitShutdown()
-    waitForPartitionState(tp, KafkaController.InitialControllerEpoch, controllerId, LeaderAndIsr.initialLeaderEpoch
+ 1,
-      "failed to get expected partition state upon broker shutdown")
-    servers(otherBrokerId).startup()
-    TestUtils.waitUntilTrue(() => zkUtils.getInSyncReplicasForPartition(tp.topic, tp.partition).toSet
== assignment(tp.partition).toSet, "restarted broker failed to join in-sync replicas")
-    zkUtils.createPersistentPath(ZkUtils.PreferredReplicaLeaderElectionPath, ZkUtils.preferredReplicaLeaderElectionZkData(Set(tp)))
-    TestUtils.waitUntilTrue(() => !zkUtils.pathExists(ZkUtils.PreferredReplicaLeaderElectionPath),
-      "failed to remove preferred replica leader election path after completion")
-    waitForPartitionState(tp, KafkaController.InitialControllerEpoch, otherBrokerId, LeaderAndIsr.initialLeaderEpoch
+ 2,
-      "failed to get expected partition state upon broker startup")
+    preferredReplicaLeaderElection(controllerId, otherBroker, tp, assignment(tp.partition).toSet,
LeaderAndIsr.initialLeaderEpoch)
+  }
+
+  @Test
+  def testBackToBackPreferredReplicaLeaderElections(): Unit = {
+    servers = makeServers(2)
+    val controllerId = TestUtils.waitUntilControllerElected(zkUtils)
+    val otherBroker = servers.find(_.config.brokerId != controllerId).get
+    val tp = TopicAndPartition("t", 0)
+    val assignment = Map(tp.partition -> Seq(otherBroker.config.brokerId, controllerId))
+    TestUtils.createTopic(zkUtils, tp.topic, partitionReplicaAssignment = assignment, servers
= servers)
+    preferredReplicaLeaderElection(controllerId, otherBroker, tp, assignment(tp.partition).toSet,
LeaderAndIsr.initialLeaderEpoch)
+    preferredReplicaLeaderElection(controllerId, otherBroker, tp, assignment(tp.partition).toSet,
LeaderAndIsr.initialLeaderEpoch + 2)
   }
 
   @Test
@@ -291,6 +293,21 @@ class ControllerIntegrationTest extends ZooKeeperTestHarness {
     }, "failed to get expected partition state after entire isr went offline")
   }
 
+  private def preferredReplicaLeaderElection(controllerId: Int, otherBroker: KafkaServer,
tp: TopicAndPartition,
+                                             replicas: Set[Int], leaderEpoch: Int): Unit
= {
+    otherBroker.shutdown()
+    otherBroker.awaitShutdown()
+    waitForPartitionState(tp, KafkaController.InitialControllerEpoch, controllerId, leaderEpoch
+ 1,
+      "failed to get expected partition state upon broker shutdown")
+    otherBroker.startup()
+    TestUtils.waitUntilTrue(() => zkUtils.getInSyncReplicasForPartition(tp.topic, tp.partition).toSet
== replicas, "restarted broker failed to join in-sync replicas")
+    zkUtils.createPersistentPath(ZkUtils.PreferredReplicaLeaderElectionPath, ZkUtils.preferredReplicaLeaderElectionZkData(Set(tp)))
+    TestUtils.waitUntilTrue(() => !zkUtils.pathExists(ZkUtils.PreferredReplicaLeaderElectionPath),
+      "failed to remove preferred replica leader election path after completion")
+    waitForPartitionState(tp, KafkaController.InitialControllerEpoch, otherBroker.config.brokerId,
leaderEpoch + 2,
+      "failed to get expected partition state upon broker startup")
+  }
+
   private def waitUntilControllerEpoch(epoch: Int, message: String): Unit = {
     TestUtils.waitUntilTrue(() => zkUtils.readDataMaybeNull(ZkUtils.ControllerEpochPath)._1.map(_.toInt)
== Some(epoch), message)
   }


Mime
View raw message