Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions core/src/main/scala/kafka/admin/TopicCommand.scala
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
56 changes: 56 additions & 0 deletions core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala
Original file line number Diff line number Diff line change
@@ -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.
*/
Comment thread
jsancio marked this conversation as resolved.
Outdated
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.jdk.CollectionConverters._

class TopicCommandTest {
@Test
def testIsNotUnderReplicatedWhenAdding(): Unit = {

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This test fails against trunk

kafka.admin.TopicCommandTest > testIsNotUnderReplicatedWhenAdding FAILED
    java.lang.AssertionError
        at org.junit.Assert.fail(Assert.java:87)
        at org.junit.Assert.assertTrue(Assert.java:42)
        at org.junit.Assert.assertFalse(Assert.java:65)
        at org.junit.Assert.assertFalse(Assert.java:75)
        at kafka.admin.TopicCommandTest.testIsNotUnderReplicatedWhenAdding(TopicCommandTest.scala:54)

1 test completed, 1 failed

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)
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ import org.scalatest.Assertions.intercept

import scala.util.Random

class TopicCommandTest extends ZooKeeperTestHarness with Logging with RackAwareTest {
class TopicCommandWithZKClientTest extends ZooKeeperTestHarness with Logging with RackAwareTest {

private var topicService: ZookeeperTopicService = _
private var testTopicName: String = _
Expand Down