MINOR: Eliminate warnings for AdminUtils

core/src/test/scala/unit/kafka/admin/AdminRackAwareTest.scala

@@ -18,7 +18,8 @@ package kafka.admin
 
 import kafka.utils.{CoreUtils, Logging}
 import org.apache.kafka.admin.{AdminUtils, BrokerMetadata}
-import org.apache.kafka.common.errors.InvalidReplicationFactorException
+import org.apache.kafka.common.errors.{InvalidPartitionsException, InvalidReplicationFactorException}
+import org.apache.kafka.server.common.AdminOperationException
 import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.Test
 
@@ -225,4 +226,26 @@ class AdminRackAwareTest extends RackAwareTest with Logging {
     val actualAssignment = CoreUtils.replicaToBrokerAssignmentAsScala(AdminUtils.assignReplicasToBrokers(brokerMetadatas, 10, 3, 0, -1))
     assertEquals(expectedAssignment, actualAssignment)
   }
+
+  @Test
+  def testAssignReplicasToBrokersWithInvalidParameters(): Unit = {
+    val rackMap = Map(0 -> "rack1", 1 -> "rack3", 2 -> "rack3", 3 -> "rack2", 4 -> null)
+    val brokerMetadatas = toBrokerMetadata(rackMap)
+
+    // test 0 partition
+    assertThrows(classOf[InvalidPartitionsException],
+      () => AdminUtils.assignReplicasToBrokers(brokerMetadatas, 0, 0, -1, -1))
+
+    // test 0 replication factor
+    assertThrows(classOf[InvalidReplicationFactorException],
+      () => AdminUtils.assignReplicasToBrokers(brokerMetadatas, 10, 0, -1, -1))
+
+    // test wrong replication factor
+    assertThrows(classOf[InvalidReplicationFactorException],
+      () => AdminUtils.assignReplicasToBrokers(brokerMetadatas, 10, brokerMetadatas.size() + 1, -1, -1))
+
+    // test wrong brokerMetadatas
+    assertThrows(classOf[AdminOperationException],
+      () => AdminUtils.assignReplicasToBrokers(brokerMetadatas, 10, brokerMetadatas.size(), -1, -1))
+  }
 }

core/src/test/scala/unit/kafka/admin/RackAwareTest.scala

@@ -81,7 +81,7 @@ trait RackAwareTest {
 
   def toBrokerMetadata(rackMap: Map[Int, String], brokersWithoutRack: Seq[Int] = Seq.empty): util.Collection[BrokerMetadata] = {
     val res = rackMap.toSeq.map { case (brokerId, rack) =>
-      new BrokerMetadata(brokerId, Optional.of(rack))
+      new BrokerMetadata(brokerId, Optional.ofNullable(rack))
     } ++ brokersWithoutRack.map { brokerId =>
       new BrokerMetadata(brokerId, Optional.empty())
     }.sortBy(_.id)

server-common/src/main/java/org/apache/kafka/admin/AdminUtils.java

@@ -130,8 +130,6 @@ public static Map<Integer, List<Integer>> assignReplicasToBrokers(Collection<Bro
             return assignReplicasToBrokersRackUnaware(nPartitions, replicationFactor, brokerMetadatas.stream().map(b -> b.id).collect(Collectors.toList()), fixedStartIndex,
                 startPartitionId);
         else {
-            if (brokerMetadatas.stream().anyMatch(b -> !b.rack.isPresent()))
-                throw new AdminOperationException("Not all brokers have rack information for replica rack aware assignment.");
             return assignReplicasToBrokersRackAware(nPartitions, replicationFactor, brokerMetadatas, fixedStartIndex,
                 startPartitionId);
         }
@@ -166,7 +164,7 @@ private static Map<Integer, List<Integer>> assignReplicasToBrokersRackAware(int
                                                                                 int fixedStartIndex,
                                                                                 int startPartitionId) {
         Map<Integer, String> brokerRackMap = new HashMap<>();
-        brokerMetadatas.forEach(m -> brokerRackMap.put(m.id, m.rack.get()));
+        brokerMetadatas.forEach(m -> brokerRackMap.put(m.id, m.rack.orElseThrow(() -> new AdminOperationException("Not all brokers have rack information for replica rack aware assignment."))));
         int numRacks = new HashSet<>(brokerRackMap.values()).size();
         List<Integer> arrangedBrokerList = getRackAlternatedBrokerList(brokerRackMap);
         int numBrokers = arrangedBrokerList.size();