MINOR: improve KRaft replica placement

Implement the existing Kafka replica placement algorithm for KRaft.
This also means implementing rack awareness.  Previously, we just chose
replicas randomly in a non-rack-aware fashion.  Also, allow replicas to
be placed on fenced brokers if there are no other choices.  This was
specified in KIP-631 but previously not implemented.

checkstyle/import-control.xml

@@ -232,6 +232,7 @@
     <allow pkg="org.apache.kafka.common.message" />
     <allow pkg="org.apache.kafka.common.metadata" />
     <allow pkg="org.apache.kafka.common.protocol" />
+    <allow pkg="org.apache.kafka.metadata" />
     <allow pkg="org.apache.kafka.server.common" />
     <allow pkg="org.apache.kafka.test" />
   </subpackage>

metadata/src/main/java/org/apache/kafka/controller/BrokerHeartbeatManager.java

@@ -25,6 +25,7 @@
 import org.slf4j.Logger;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -272,6 +273,11 @@ BrokerHeartbeatStateList unfenced() {
         return unfenced;
     }
 
+    // VisibleForTesting
+    Collection<BrokerHeartbeatState> brokers() {
+        return brokers.values();
+    }
+
     /**
      * Mark a broker as fenced.
      *
@@ -439,7 +445,7 @@ List<Integer> findStaleBrokers() {
      * @param numPartitions     The number of partitions to place.
      * @param numReplicas       The number of replicas for each partition.
      * @param idToRack          A function mapping broker id to broker rack.
-     * @param policy            The replica placement policy to use.
+     * @param placer            The replica placer to use.
      *
      * @return                  A list of replica lists.
      *
@@ -449,12 +455,10 @@ List<List<Integer>> placeReplicas(int startPartition,
                                       int numPartitions,
                                       short numReplicas,
                                       Function<Integer, Optional<String>> idToRack,
-                                      ReplicaPlacementPolicy policy) {
-        // TODO: support using fenced brokers here if necessary to get to the desired
-        // number of replicas. We probably need to add a fenced boolean in UsableBroker.
+                                      ReplicaPlacer placer) {
         Iterator<UsableBroker> iterator = new UsableBrokerIterator(
-            unfenced.iterator(), idToRack);
-        return policy.createPlacement(startPartition, numPartitions, numReplicas, iterator);
+            brokers.values().iterator(), idToRack);
+        return placer.place(startPartition, numPartitions, numReplicas, iterator);
     }
 
     static class UsableBrokerIterator implements Iterator<UsableBroker> {
@@ -482,7 +486,7 @@ public boolean hasNext() {
                 result = iterator.next();
             } while (result.shuttingDown());
             Optional<String> rack = idToRack.apply(result.id());
-            next = new UsableBroker(result.id(), rack);
+            next = new UsableBroker(result.id(), rack, result.fenced());
             return true;
         }
 

metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java

@@ -103,9 +103,9 @@ boolean check() {
     private final long sessionTimeoutNs;
 
     /**
-     * The replica placement policy to use.
+     * The replica placer to use.
      */
-    private final ReplicaPlacementPolicy placementPolicy;
+    private final ReplicaPlacer replicaPlacer;
 
     /**
      * Maps broker IDs to broker registrations.
@@ -127,12 +127,12 @@ boolean check() {
                           Time time,
                           SnapshotRegistry snapshotRegistry,
                           long sessionTimeoutNs,
-                          ReplicaPlacementPolicy placementPolicy) {
+                          ReplicaPlacer replicaPlacer) {
         this.logContext = logContext;
         this.log = logContext.logger(ClusterControlManager.class);
         this.time = time;
         this.sessionTimeoutNs = sessionTimeoutNs;
-        this.placementPolicy = placementPolicy;
+        this.replicaPlacer = replicaPlacer;
         this.brokerRegistrations = new TimelineHashMap<>(snapshotRegistry, 0);
         this.heartbeatManager = null;
         this.readyBrokersFuture = Optional.empty();
@@ -317,7 +317,7 @@ public List<List<Integer>> placeReplicas(int startPartition,
             throw new RuntimeException("ClusterControlManager is not active.");
         }
         return heartbeatManager.placeReplicas(startPartition, numPartitions, numReplicas,
-            id -> brokerRegistrations.get(id).rack(), placementPolicy);
+            id -> brokerRegistrations.get(id).rack(), replicaPlacer);
     }
 
     public boolean unfenced(int brokerId) {

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java

@@ -122,8 +122,7 @@ static public class Builder {
         private Map<String, VersionRange> supportedFeatures = Collections.emptyMap();
         private short defaultReplicationFactor = 3;
         private int defaultNumPartitions = 1;
-        private ReplicaPlacementPolicy replicaPlacementPolicy =
-            new SimpleReplicaPlacementPolicy(new Random());
+        private ReplicaPlacer replicaPlacer = new StripedReplicaPlacer(new Random());
         private Function<Long, SnapshotWriter> snapshotWriterBuilder;
         private SnapshotReader snapshotReader;
         private long sessionTimeoutNs = NANOSECONDS.convert(18, TimeUnit.SECONDS);
@@ -173,8 +172,8 @@ public Builder setDefaultNumPartitions(int defaultNumPartitions) {
             return this;
         }
 
-        public Builder setReplicaPlacementPolicy(ReplicaPlacementPolicy replicaPlacementPolicy) {
-            this.replicaPlacementPolicy = replicaPlacementPolicy;
+        public Builder setReplicaPlacer(ReplicaPlacer replicaPlacer) {
+            this.replicaPlacer = replicaPlacer;
             return this;
         }
 
@@ -224,7 +223,7 @@ public QuorumController build() throws Exception {
                 queue = new KafkaEventQueue(time, logContext, threadNamePrefix);
                 return new QuorumController(logContext, nodeId, queue, time, configDefs,
                     logManager, supportedFeatures, defaultReplicationFactor,
-                    defaultNumPartitions, replicaPlacementPolicy, snapshotWriterBuilder,
+                    defaultNumPartitions, replicaPlacer, snapshotWriterBuilder,
                     snapshotReader, sessionTimeoutNs, controllerMetrics);
             } catch (Exception e) {
                 Utils.closeQuietly(queue, "event queue");
@@ -906,7 +905,7 @@ private QuorumController(LogContext logContext,
                              Map<String, VersionRange> supportedFeatures,
                              short defaultReplicationFactor,
                              int defaultNumPartitions,
-                             ReplicaPlacementPolicy replicaPlacementPolicy,
+                             ReplicaPlacer replicaPlacer,
                              Function<Long, SnapshotWriter> snapshotWriterBuilder,
                              SnapshotReader snapshotReader,
                              long sessionTimeoutNs,
@@ -923,7 +922,7 @@ private QuorumController(LogContext logContext,
             snapshotRegistry, configDefs);
         this.clientQuotaControlManager = new ClientQuotaControlManager(snapshotRegistry);
         this.clusterControl = new ClusterControlManager(logContext, time,
-            snapshotRegistry, sessionTimeoutNs, replicaPlacementPolicy);
+            snapshotRegistry, sessionTimeoutNs, replicaPlacer);
         this.featureControl = new FeatureControlManager(supportedFeatures, snapshotRegistry);
         this.snapshotGeneratorManager = new SnapshotGeneratorManager(snapshotWriterBuilder);
         this.replicationControl = new ReplicationControlManager(snapshotRegistry,

metadata/src/main/java/org/apache/kafka/controller/ReplicaPlacementPolicy.java → metadata/src/main/java/org/apache/kafka/controller/ReplicaPlacer.java

@@ -28,7 +28,7 @@
  * The interface which a Kafka replica placement policy must implement.
  */
 @InterfaceStability.Unstable
-interface ReplicaPlacementPolicy {
+interface ReplicaPlacer {
     /**
      * Create a new replica placement.
      *
@@ -42,9 +42,9 @@ interface ReplicaPlacementPolicy {
      *
      * @throws InvalidReplicationFactorException    If too many replicas were requested.
      */
-    List<List<Integer>> createPlacement(int startPartition,
-                                        int numPartitions,
-                                        short numReplicas,
-                                        Iterator<UsableBroker> iterator)
+    List<List<Integer>> place(int startPartition,
+                              int numPartitions,
+                              short numReplicas,
+                              Iterator<UsableBroker> iterator)
         throws InvalidReplicationFactorException;
 }

metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java

@@ -510,7 +510,7 @@ private ApiError createTopic(CreatableTopic topic,
             } catch (InvalidReplicationFactorException e) {
                 return new ApiError(Errors.INVALID_REPLICATION_FACTOR,
                     "Unable to replicate the partition " + replicationFactor +
-                        " times: " + e.getMessage());
+                        " time(s): " + e.getMessage());
             }
         }
         Uuid topicId = Uuid.randomUuid();