MINOR: Partition is under reassignment when adding and removing

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(_.intValue).toSet ++ ra.addingReplicas.asScala.map(_.intValue).toSet
+      allReplicaIds.exists(changingReplicaIds.contains)
     }
 
     reassignment match {

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)
+  }
+}

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)

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 = _