Repository: kafka Updated Branches: refs/heads/trunk 448c1a411 -> b8c3b7c96
KAFKA-4596; Throttled reassignment should support partial JSON file Fixes a logic error in the Reassignment process which throws an exception if you don't rebalance all partitions. Author: Ben Stopford <[email protected]> Reviewers: Ismael Juma <[email protected]> Closes #2399 from benstopford/KAFKA-4596 Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/b8c3b7c9 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/b8c3b7c9 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/b8c3b7c9 Branch: refs/heads/trunk Commit: b8c3b7c964c6167be7f8ad72c96733dfa0d648a2 Parents: 448c1a4 Author: Ben Stopford <[email protected]> Authored: Wed Jan 25 12:23:57 2017 +0000 Committer: Ismael Juma <[email protected]> Committed: Wed Jan 25 12:38:47 2017 +0000 ---------------------------------------------------------------------- .../kafka/admin/ReassignPartitionsCommand.scala | 13 +-- .../admin/ReassignPartitionsClusterTest.scala | 91 ++++++++++++++++++-- .../admin/ReassignPartitionsCommandTest.scala | 33 +++++++ 3 files changed, 124 insertions(+), 13 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/b8c3b7c9/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala b/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala index dc707e5..4e7b4e0 100755 --- a/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala +++ b/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala @@ -360,16 +360,17 @@ class ReassignPartitionsCommand(zkUtils: ZkUtils, proposedAssignment: Map[TopicA } private def postRebalanceReplicasThatMoved(existing: Map[TopicAndPartition, Seq[Int]], proposed: Map[TopicAndPartition, Seq[Int]]): Map[TopicAndPartition, Seq[Int]] = { - //For each partition in the proposed list, filter out any replicas that exist now (i.e. are in the proposed list and hence are not moving) - existing.map { case (tp, current) => - tp -> (proposed(tp).toSet -- current).toSeq + //For each partition in the proposed list, filter out any replicas that exist now, and hence aren't being moved. + proposed.map { case (tp, proposedReplicas) => + tp -> (proposedReplicas.toSet -- existing(tp)).toSeq } } private def preRebalanceReplicaForMovingPartitions(existing: Map[TopicAndPartition, Seq[Int]], proposed: Map[TopicAndPartition, Seq[Int]]): Map[TopicAndPartition, Seq[Int]] = { - //Throttle all existing replicas (as any one might be a leader). So just filter out those which aren't moving - existing.filter { case (tp, current) => - (proposed(tp).toSet -- current).nonEmpty + def moving(before: Seq[Int], after: Seq[Int]) = (after.toSet -- before.toSet).nonEmpty + //For any moving partition, throttle all the original (pre move) replicas (as any one might be a leader) + existing.filter { case (tp, preMoveReplicas) => + proposed.contains(tp) && moving(preMoveReplicas, proposed(tp)) } } http://git-wip-us.apache.org/repos/asf/kafka/blob/b8c3b7c9/core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala b/core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala index 95ee5d3..c211c24 100644 --- a/core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala @@ -16,12 +16,13 @@ import kafka.common.{AdminCommandFailedException, TopicAndPartition} import kafka.server.{KafkaConfig, KafkaServer} import kafka.utils.TestUtils._ import kafka.utils.ZkUtils._ -import kafka.utils.{CoreUtils, Logging, ZkUtils} +import kafka.utils.{CoreUtils, Logging, TestUtils, ZkUtils} import kafka.zk.ZooKeeperTestHarness import org.junit.Assert.{assertEquals, assertTrue} import org.junit.{After, Before, Test} import kafka.admin.ReplicationQuotaUtils._ -import scala.collection.Seq + +import scala.collection.{Map, Seq} class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging { @@ -58,7 +59,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging { waitForReassignmentToComplete() //Then the replica should be on 101 - assertEquals(zkUtils.getPartitionAssignmentForTopics(Seq(topicName)).get(topicName).get(partition), Seq(101)) + assertEquals(Seq(101), zkUtils.getPartitionAssignmentForTopics(Seq(topicName)).get(topicName).get(partition)) } @Test @@ -79,7 +80,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging { //Then the replicas should span all three brokers val actual = zkUtils.getPartitionAssignmentForTopics(Seq(topicName))(topicName) - assertEquals(actual.values.flatten.toSeq.distinct.sorted, Seq(100, 101, 102)) + assertEquals(Seq(100, 101, 102), actual.values.flatten.toSeq.distinct.sorted) } @Test @@ -100,7 +101,43 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging { //Then replicas should only span the first two brokers val actual = zkUtils.getPartitionAssignmentForTopics(Seq(topicName))(topicName) - assertEquals(actual.values.flatten.toSeq.distinct.sorted, Seq(100, 101)) + assertEquals(Seq(100, 101), actual.values.flatten.toSeq.distinct.sorted) + } + + @Test + def shouldMoveSubsetOfPartitions() { + //Given partitions on 3 of 3 brokers + val brokers = Array(100, 101, 102) + startBrokers(brokers) + createTopic(zkUtils, "topic1", Map( + 0 -> Seq(100, 101), + 1 -> Seq(101, 102), + 2 -> Seq(102, 100) + ), servers = servers) + createTopic(zkUtils, "topic2", Map( + 0 -> Seq(100, 101), + 1 -> Seq(101, 102), + 2 -> Seq(102, 100) + ), servers = servers) + + val proposed: Map[TopicAndPartition, Seq[Int]] = Map( + TopicAndPartition("topic1", 0) -> Seq(100, 102), + TopicAndPartition("topic1", 2) -> Seq(100, 102), + TopicAndPartition("topic2", 2) -> Seq(100, 102) + ) + + //When rebalancing + ReassignPartitionsCommand.executeAssignment(zkUtils, ZkUtils.formatAsReassignmentJson(proposed)) + waitForReassignmentToComplete() + + //Then the proposed changes should have been made + val actual = zkUtils.getPartitionAssignmentForTopics(Seq("topic1", "topic2")) + assertEquals(Seq(100, 102), actual("topic1")(0))//changed + assertEquals(Seq(101, 102), actual("topic1")(1)) + assertEquals(Seq(100, 102), actual("topic1")(2))//changed + assertEquals(Seq(100, 101), actual("topic2")(0)) + assertEquals(Seq(101, 102), actual("topic2")(1)) + assertEquals(Seq(100, 102), actual("topic2")(2))//changed } @Test @@ -135,7 +172,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging { //Check move occurred val actual = zkUtils.getPartitionAssignmentForTopics(Seq(topicName))(topicName) - assertEquals(actual.values.flatten.toSeq.distinct.sorted, Seq(101, 102)) + assertEquals(Seq(101, 102), actual.values.flatten.toSeq.distinct.sorted) //Then command should have taken longer than the throttle rate assertTrue(s"Expected replication to be > ${expectedDurationSecs * 0.9 * 1000} but was $took", took > expectedDurationSecs * 0.9 * 1000) @@ -233,7 +270,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging { //Check move occurred val actual = zkUtils.getPartitionAssignmentForTopics(Seq(topicName))(topicName) - assertEquals(actual.values.flatten.toSeq.distinct.sorted, Seq(101, 102)) + assertEquals(Seq(101, 102), actual.values.flatten.toSeq.distinct.sorted) } @Test(expected = classOf[AdminCommandFailedException]) @@ -246,6 +283,46 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging { ReassignPartitionsCommand.executeAssignment(zkUtils, s"""{"version":1,"partitions":[{"topic":"$topicName","partition":1,"replicas":[101]}]}""") } + @Test + def shouldPerformThrottledReassignmentOverVariousTopics() { + val throttle = 1000L + + //Given four brokers + servers = TestUtils.createBrokerConfigs(4, zkConnect, false).map(conf => TestUtils.createServer(KafkaConfig.fromProps(conf))) + + //With up several small topics + createTopic(zkUtils, "orders", Map(0 -> List(0, 1, 2), 1 -> List(0, 1, 2)), servers) + createTopic(zkUtils, "payments", Map(0 -> List(0, 1), 1 -> List(0, 1)), servers) + createTopic(zkUtils, "deliveries", Map(0 -> List(0)), servers) + createTopic(zkUtils, "customers", Map(0 -> List(0), 1 -> List(1), 2 -> List(2), 3 -> List(3)), servers) + + //Define a move for some of them + val move = Map( + TopicAndPartition("orders", 0) -> Seq(0, 2, 3),//moves + TopicAndPartition("orders", 1) -> Seq(0, 1, 2),//stays + TopicAndPartition("payments", 1) -> Seq(1, 2), //only define one partition as moving + TopicAndPartition("deliveries", 0) -> Seq(1, 2) //increase replication factor + ) + + //When we run a throttled reassignment + new ReassignPartitionsCommand(zkUtils, move).reassignPartitions(throttle) + + waitForReassignmentToComplete() + + //Check moved replicas did move + assertEquals(Seq(0, 2, 3), zkUtils.getReplicasForPartition("orders", 0)) + assertEquals(Seq(0, 1, 2), zkUtils.getReplicasForPartition("orders", 1)) + assertEquals(Seq(1, 2), zkUtils.getReplicasForPartition("payments", 1)) + assertEquals(Seq(1, 2), zkUtils.getReplicasForPartition("deliveries", 0)) + + //Check untouched replicas are still there + assertEquals(Seq(0, 1), zkUtils.getReplicasForPartition("payments", 0)) + assertEquals(Seq(0), zkUtils.getReplicasForPartition("customers", 0)) + assertEquals(Seq(1), zkUtils.getReplicasForPartition("customers", 1)) + assertEquals(Seq(2), zkUtils.getReplicasForPartition("customers", 2)) + assertEquals(Seq(3), zkUtils.getReplicasForPartition("customers", 3)) + } + def waitForReassignmentToComplete() { waitUntilTrue(() => !zkUtils.pathExists(ReassignPartitionsPath), s"Znode ${ZkUtils.ReassignPartitionsPath} wasn't deleted") } http://git-wip-us.apache.org/repos/asf/kafka/blob/b8c3b7c9/core/src/test/scala/unit/kafka/admin/ReassignPartitionsCommandTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/admin/ReassignPartitionsCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ReassignPartitionsCommandTest.scala index 924daf8..5ecc19b 100644 --- a/core/src/test/scala/unit/kafka/admin/ReassignPartitionsCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ReassignPartitionsCommandTest.scala @@ -29,6 +29,7 @@ import org.easymock.EasyMock._ import org.easymock.{Capture, CaptureType, EasyMock} import org.junit.{Before, Test} import org.junit.Assert.{assertEquals, assertNull, fail} + import scala.collection.{Seq, mutable} import scala.collection.JavaConversions._ @@ -58,6 +59,38 @@ class ReassignPartitionsCommandTest extends Logging { } @Test + def shouldFindMovingReplicasWhenProposedIsSubsetOfExisting() { + val assigner = new ReassignPartitionsCommand(null, null) + + //Given we have more existing partitions than we are proposing + val existingSuperset = Map( + TopicAndPartition("topic1", 0) -> Seq(100, 101), + TopicAndPartition("topic1", 1) -> Seq(100, 102), + TopicAndPartition("topic1", 2) -> Seq(100, 101), + TopicAndPartition("topic2", 0) -> Seq(100, 101, 102), + TopicAndPartition("topic3", 0) -> Seq(100, 101, 102) + ) + val proposedSubset = Map( + TopicAndPartition("topic1", 0) -> Seq(101, 102), + TopicAndPartition("topic1", 1) -> Seq(102), + TopicAndPartition("topic1", 2) -> Seq(100, 101, 102) + ) + + val mock = new TestAdminUtils { + override def changeTopicConfig(zkUtils: ZkUtils, topic: String, configChange: Properties): Unit = { + assertEquals("0:102,2:102", configChange.get(FollowerReplicationThrottledReplicasProp)) + assertEquals("0:100,0:101,2:100,2:101", configChange.get(LeaderReplicationThrottledReplicasProp)) + assertEquals("topic1", topic) + calls += 1 + } + } + + //Then replicas should assign correctly (based on the proposed map) + assigner.assignThrottledReplicas(existingSuperset, proposedSubset, mock) + assertEquals(1, calls) + } + + @Test def shouldFindMovingReplicasMultiplePartitions() { val control = TopicAndPartition("topic1", 2) -> Seq(100, 102) val assigner = new ReassignPartitionsCommand(null, null)
