From eb0bc802928ad0d4a44966a42a0abacf69d9358c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Armando=20Garc=C3=ADa=20Sancio?= Date: Thu, 26 Mar 2020 08:25:52 -0700 Subject: [PATCH 1/4] MINOR: Parition is under reassignment when adding and removing A partition is under reassignment if the either the set of adding replicas or set removing replicas is non-empty. Fix the test assertion such that it prints stdout on failure. --- core/src/main/scala/kafka/admin/TopicCommand.scala | 6 +++--- .../unit/kafka/admin/TopicCommandWithAdminClientTest.scala | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala index 9c2c22369e46f..6abde0221d508 100755 --- a/core/src/main/scala/kafka/admin/TopicCommand.scala +++ b/core/src/main/scala/kafka/admin/TopicCommand.scala @@ -563,10 +563,10 @@ object TopicCommand extends Logging { // It is possible for a reassignment to complete between the time we have fetched its state and the time // we fetch partition metadata. In ths case, we ignore the reassignment when determining replication factor. def isReassignmentInProgress(ra: PartitionReassignment): Boolean = { - // Reassignment is still in progress as long as the removing replicas are still present + // Reassignment is still in progress as long as the removing and adding replicas are still present val allReplicaIds = tpi.replicas.asScala.map(_.id).toSet - val removingReplicaIds = ra.removingReplicas.asScala.map(Int.unbox).toSet - allReplicaIds.exists(removingReplicaIds.contains) + val changingReplicaIds = ra.removingReplicas.asScala.map(Int.unbox).toSet ++ ra.addingReplicas.asScala.map(Int.unbox).toSet + allReplicaIds.exists(changingReplicaIds.contains) } reassignment match { diff --git a/core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala b/core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala index 3f95038e7d96a..928ff8db14137 100644 --- a/core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala +++ b/core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala @@ -682,7 +682,7 @@ class TopicCommandWithAdminClientTest extends KafkaServerTestHarness with Loggin val underReplicatedOutput = TestUtils.grabConsoleOutput( topicService.describeTopic(new TopicCommandOptions(Array("--under-replicated-partitions")))) - assertEquals("--under-replicated-partitions shouldn't return anything", "", underReplicatedOutput) + assertEquals(s"--under-replicated-partitions shouldn't return anything: '$underReplicatedOutput'", "", underReplicatedOutput) TestUtils.removeReplicationThrottleForPartitions(adminClient, brokerIds, Set(tp)) TestUtils.waitForAllReassignmentsToComplete(adminClient) From a7d69ce0899117265666d65baa81555538c9a410 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Armando=20Garc=C3=ADa=20Sancio?= Date: Mon, 30 Mar 2020 19:58:18 -0700 Subject: [PATCH 2/4] Use intValue instead of unbox --- core/src/main/scala/kafka/admin/TopicCommand.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala index 6abde0221d508..86dc84e6b58dd 100755 --- a/core/src/main/scala/kafka/admin/TopicCommand.scala +++ b/core/src/main/scala/kafka/admin/TopicCommand.scala @@ -565,7 +565,7 @@ object TopicCommand extends Logging { def isReassignmentInProgress(ra: PartitionReassignment): Boolean = { // Reassignment is still in progress as long as the removing and adding replicas are still present val allReplicaIds = tpi.replicas.asScala.map(_.id).toSet - val changingReplicaIds = ra.removingReplicas.asScala.map(Int.unbox).toSet ++ ra.addingReplicas.asScala.map(Int.unbox).toSet + val changingReplicaIds = ra.removingReplicas.asScala.map(_.intValue).toSet ++ ra.addingReplicas.asScala.map(_.intValue).toSet allReplicaIds.exists(changingReplicaIds.contains) } From f276d286e7c15679430eaee5c4504e7c213694b5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Armando=20Garc=C3=ADa=20Sancio?= Date: Tue, 31 Mar 2020 14:15:38 -0700 Subject: [PATCH 3/4] Unit test for false URP when adding --- .../unit/kafka/admin/TopicCommandTest.scala | 56 +++++++++++++++++++ .../admin/TopicCommandWithZKClientTest.scala | 2 +- 2 files changed, 57 insertions(+), 1 deletion(-) create mode 100644 core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala diff --git a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala new file mode 100644 index 0000000000000..2414046dc187e --- /dev/null +++ b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala @@ -0,0 +1,56 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package kafka.admin + +import kafka.admin.TopicCommand.PartitionDescription +import org.apache.kafka.clients.admin.PartitionReassignment +import org.apache.kafka.common.Node +import org.apache.kafka.common.TopicPartitionInfo +import org.junit.Assert._ +import org.junit.Test +import scala.collection.JavaConverters._ + +final class TopicCommandTest { + @Test + def testIsNotUnderReplicatedWhenAdding(): Unit = { + val replicaIds = List(1, 2) + val replicas = replicaIds.map { id => + new Node(id, "localhost", 9090 + id) + } + + val partitionDescription = PartitionDescription( + "test-topic", + new TopicPartitionInfo( + 0, + new Node(1, "localhost", 9091), + replicas.asJava, + List(new Node(1, "localhost", 9091)).asJava + ), + None, + false, + Some( + new PartitionReassignment( + replicaIds.map(id => id: java.lang.Integer).asJava, + List(2: java.lang.Integer).asJava, + List.empty.asJava + ) + ) + ) + + assertFalse(partitionDescription.isUnderReplicated) + } +} diff --git a/core/src/test/scala/unit/kafka/admin/TopicCommandWithZKClientTest.scala b/core/src/test/scala/unit/kafka/admin/TopicCommandWithZKClientTest.scala index db9eaed386cd8..256be6a00e981 100644 --- a/core/src/test/scala/unit/kafka/admin/TopicCommandWithZKClientTest.scala +++ b/core/src/test/scala/unit/kafka/admin/TopicCommandWithZKClientTest.scala @@ -31,7 +31,7 @@ import org.scalatest.Assertions.intercept import scala.util.Random -class TopicCommandTest extends ZooKeeperTestHarness with Logging with RackAwareTest { +final class TopicCommandWithZKClientTest extends ZooKeeperTestHarness with Logging with RackAwareTest { private var topicService: ZookeeperTopicService = _ private var testTopicName: String = _ From e4e2760c25867cd26ab27f12c45cd95ce0101243 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Armando=20Garc=C3=ADa=20Sancio?= Date: Fri, 24 Apr 2020 15:00:26 -0700 Subject: [PATCH 4/4] Use CollectionConverters --- core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala | 4 ++-- .../scala/unit/kafka/admin/TopicCommandWithZKClientTest.scala | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala index 2414046dc187e..9ea8a80230222 100644 --- a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala @@ -22,9 +22,9 @@ import org.apache.kafka.common.Node import org.apache.kafka.common.TopicPartitionInfo import org.junit.Assert._ import org.junit.Test -import scala.collection.JavaConverters._ +import scala.jdk.CollectionConverters._ -final class TopicCommandTest { +class TopicCommandTest { @Test def testIsNotUnderReplicatedWhenAdding(): Unit = { val replicaIds = List(1, 2) diff --git a/core/src/test/scala/unit/kafka/admin/TopicCommandWithZKClientTest.scala b/core/src/test/scala/unit/kafka/admin/TopicCommandWithZKClientTest.scala index 256be6a00e981..64233b5229597 100644 --- a/core/src/test/scala/unit/kafka/admin/TopicCommandWithZKClientTest.scala +++ b/core/src/test/scala/unit/kafka/admin/TopicCommandWithZKClientTest.scala @@ -31,7 +31,7 @@ import org.scalatest.Assertions.intercept import scala.util.Random -final class TopicCommandWithZKClientTest extends ZooKeeperTestHarness with Logging with RackAwareTest { +class TopicCommandWithZKClientTest extends ZooKeeperTestHarness with Logging with RackAwareTest { private var topicService: ZookeeperTopicService = _ private var testTopicName: String = _