KAFKA-20897: Add broker pods and canary partition placement isolation (KIP-1095)

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -1811,7 +1811,7 @@ static Cluster parseDescribeClusterResponse(DescribeClusterResponseData response
         List<Node> nodes = new ArrayList<>();
         Node controllerNode = null;
         for (DescribeClusterResponseData.DescribeClusterBroker node : response.brokers()) {
-            Node newNode = new Node(node.brokerId(), node.host(), node.port(), node.rack());
+            Node newNode = new Node(node.brokerId(), node.host(), node.port(), node.rack(), node.pod());
             nodes.add(newNode);
             if (node.brokerId() == response.controllerId()) {
                 controllerNode = newNode;

clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupCoordinatorNode.java

@@ -30,7 +30,7 @@
  */
 public class GroupCoordinatorNode extends Node {
     public GroupCoordinatorNode(int id, String host, int port) {
-        super(GroupCoordinatorNode.validateId(id), host, port, null, false, "+" + id);
+        super(GroupCoordinatorNode.validateId(id), host, port, null, null, false, "+" + id);
     }
 
     private static int validateId(int id) {

clients/src/main/java/org/apache/kafka/common/Node.java

@@ -34,29 +34,35 @@ public class Node {
     private final String host;
     private final int port;
     private final String rack;
+    private final String pod;
     private final boolean isFenced;
 
     // Cache hashCode as it is called in performance sensitive parts of the code (e.g. RecordAccumulator.ready)
     private Integer hash;
 
     public Node(int id, String host, int port) {
-        this(id, host, port, null, false);
+        this(id, host, port, null, null, false);
     }
 
     public Node(int id, String host, int port, String rack) {
-        this(id, host, port, rack, false);
+        this(id, host, port, rack, null, false);
     }
 
-    public Node(int id, String host, int port, String rack, boolean isFenced) {
-        this(id, host, port, rack, isFenced, Integer.toString(id));
+    public Node(int id, String host, int port, String rack, String pod) {
+        this(id, host, port, rack, pod, false);
     }
 
-    protected Node(int id, String host, int port, String rack, boolean isFenced, String idString) {
+    public Node(int id, String host, int port, String rack, String pod, boolean isFenced) {
+        this(id, host, port, rack, pod, isFenced, Integer.toString(id));
+    }
+
+    protected Node(int id, String host, int port, String rack, String pod, boolean isFenced, String idString) {
         this.id = id;
         this.idString = idString;
         this.host = host;
         this.port = port;
         this.rack = rack;
+        this.pod = pod;
         this.isFenced = isFenced;
     }
 
@@ -127,6 +133,20 @@ public boolean isFenced() {
         return isFenced;
     }
 
+    /**
+     * True if this node has a defined pod
+     */
+    public boolean hasPod() {
+        return pod != null;
+    }
+
+    /**
+     * @return the pod of the node
+     */
+    public String pod() {
+        return pod;
+    }
+
     @Override
     public int hashCode() {
         Integer h = this.hash;
@@ -135,6 +155,7 @@ public int hashCode() {
             result = 31 * result + id;
             result = 31 * result + port;
             result = 31 * result + ((rack == null) ? 0 : rack.hashCode());
+            result = 31 * result + ((pod == null) ? 0 : pod.hashCode());
             result = 31 * result + Objects.hashCode(isFenced);
             this.hash = result;
             return result;
@@ -154,11 +175,12 @@ public boolean equals(Object obj) {
             port == other.port &&
             Objects.equals(host, other.host) &&
             Objects.equals(rack, other.rack) &&
+            Objects.equals(pod, other.pod) &&
             Objects.equals(isFenced, other.isFenced);
     }
 
     @Override
     public String toString() {
-        return host + ":" + port + " (id: " + idString + " rack: " + rack + " isFenced: " + isFenced + ")";
+        return host + ":" + port + " (id: " + idString + " rack: " + rack + " pod: " + pod + " isFenced: " + isFenced + ")";
     }
 }

clients/src/main/java/org/apache/kafka/common/requests/DescribeClusterResponse.java

@@ -38,7 +38,7 @@ public DescribeClusterResponse(DescribeClusterResponseData data) {
 
     public Map<Integer, Node> nodes() {
         return data.brokers().valuesList().stream()
-            .map(b -> new Node(b.brokerId(), b.host(), b.port(), b.rack(), b.isFenced()))
+            .map(b -> new Node(b.brokerId(), b.host(), b.port(), b.rack(), b.pod(), b.isFenced()))
             .collect(Collectors.toMap(Node::id, Function.identity()));
     }
 

clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java

@@ -442,7 +442,7 @@ private static class Holder {
         }
 
         private Map<Integer, Node> createBrokers(MetadataResponseData data) {
-            return data.brokers().valuesList().stream().map(b -> new Node(b.nodeId(), b.host(), b.port(), b.rack()))
+            return data.brokers().valuesList().stream().map(b -> new Node(b.nodeId(), b.host(), b.port(), b.rack(), b.pod()))
                     .collect(Collectors.toMap(Node::id, Function.identity()));
         }
 
@@ -503,7 +503,8 @@ public static MetadataResponse prepareResponse(boolean hasReliableEpoch,
                 .setNodeId(broker.id())
                 .setHost(broker.host())
                 .setPort(broker.port())
-                .setRack(broker.rack()))
+                .setRack(broker.rack())
+                .setPod(broker.pod()))
         );
 
         responseData.setClusterId(clusterId);

clients/src/main/resources/common/message/BrokerRegistrationRequest.json

@@ -60,6 +60,8 @@
     { "name": "LogDirs", "type":  "[]uuid", "versions":  "2+",
       "about": "Log directories configured in this broker which are available.", "ignorable": true },
     { "name": "PreviousBrokerEpoch", "type": "int64", "versions": "3+", "default": "-1", "ignorable": true,
-      "about": "The epoch before a clean shutdown." }
+      "about": "The epoch before a clean shutdown." },
+    { "name": "Pod", "type": "string", "versions": "0+", "nullableVersions": "0+", "taggedVersions": "0+", "tag": 0, "default": "null",
+      "about": "The Pod which this broker is in." }
   ]
 }

clients/src/main/resources/common/message/DescribeClusterResponse.json

@@ -48,7 +48,9 @@
       { "name": "Rack", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
         "about": "The rack of the broker, or null if it has not been assigned to a rack." },
       { "name": "IsFenced", "type": "bool", "versions": "2+",
-        "about": "Whether the broker is fenced" }
+        "about": "Whether the broker is fenced" },
+      { "name": "Pod", "type": "string", "versions": "0+", "nullableVersions": "0+",  "default": "null",
+        "about": "The pod of the broker, or null if it has not been assigned to a pod.", "taggedVersions": "0+", "tag": 0 }
     ]},
     { "name": "ClusterAuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
       "about": "32-bit bitfield to represent authorized operations for this cluster." }

clients/src/main/resources/common/message/MetadataResponse.json

@@ -56,7 +56,9 @@
       { "name": "Port", "type": "int32", "versions": "0+",
         "about": "The broker port." },
       { "name": "Rack", "type": "string", "versions": "1+", "nullableVersions": "1+", "ignorable": true, "default": "null",
-        "about": "The rack of the broker, or null if it has not been assigned to a rack." }
+        "about": "The rack of the broker, or null if it has not been assigned to a rack." },
+      { "name": "Pod", "type": "string", "versions": "11+", "nullableVersions": "11+", "ignorable": true, "default": "null",
+        "about": "The pod of the broker, or null if it has not been assigned to a pod.", "taggedVersions": "11+", "tag": 0 }
     ]},
     { "name": "ClusterId", "type": "string", "nullableVersions": "2+", "versions": "2+", "ignorable": true, "default": "null",
       "about": "The cluster ID that responding broker belongs to." },

clients/src/test/java/org/apache/kafka/clients/MetadataTest.java

@@ -905,7 +905,8 @@ private MetadataResponseBrokerCollection buildBrokerCollection(List<Node> nodes)
                     .setNodeId(node.id())
                     .setHost(node.host())
                     .setPort(node.port())
-                    .setRack(node.rack());
+                    .setRack(node.rack())
+                    .setPod(node.pod());
             brokers.add(broker);
         }
         return brokers;

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

@@ -3002,7 +3002,8 @@ private static DescribeClusterResponse prepareDescribeClusterResponse(
                 .setHost(broker.host())
                 .setPort(broker.port())
                 .setBrokerId(broker.id())
-                .setRack(broker.rack())));
+                .setRack(broker.rack())
+                .setPod(broker.pod())));
 
         return new DescribeClusterResponse(data);
     }
@@ -10536,6 +10537,7 @@ public void testListTransactions() throws Exception {
                     .setNodeId(node.id())
                     .setPort(node.port())
                     .setRack(node.rack())
+                    .setPod(node.pod())
                 )
             );
 

connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java

@@ -928,7 +928,8 @@ private DescribeClusterResponse describeClusterResponse(Cluster cluster) {
                 .setHost(broker.host())
                 .setPort(broker.port())
                 .setBrokerId(broker.id())
-                .setRack(broker.rack())));
+                .setRack(broker.rack())
+                .setPod(broker.pod())));
 
         return new DescribeClusterResponse(data);
     }

core/src/main/scala/kafka/server/ControllerServer.scala

@@ -40,6 +40,7 @@ import org.apache.kafka.image.publisher.{ControllerRegistrationsPublisher, KRaft
 import org.apache.kafka.metadata.{KafkaConfigSchema, KRaftMetadataCache, ListenerInfo}
 import org.apache.kafka.metadata.authorizer.ClusterMetadataAuthorizer
 import org.apache.kafka.metadata.bootstrap.BootstrapMetadata
+import org.apache.kafka.metadata.placement.{PodReplicaPlacer, StripedReplicaPlacer}
 import org.apache.kafka.metadata.publisher.{AclPublisher, DelegationTokenPublisher, DynamicClientQuotaPublisher, DynamicTopicClusterQuotaPublisher, FeaturesPublisher, ScramPublisher}
 import org.apache.kafka.raft.QuorumConfig
 import org.apache.kafka.security.{CredentialProvider, DelegationTokenManager}
@@ -58,7 +59,7 @@ import org.apache.kafka.server.NodeToControllerChannelManagerImpl
 import org.apache.kafka.server.RaftControllerNodeProvider
 
 import java.util
-import java.util.{Optional, OptionalLong}
+import java.util.{Optional, OptionalLong, Random}
 import java.util.concurrent.locks.ReentrantLock
 import java.util.concurrent.{CompletableFuture, TimeUnit}
 import scala.jdk.CollectionConverters._
@@ -248,6 +249,7 @@ class ControllerServer(
           setQuorumFeatures(quorumFeatures).
           setDefaultReplicationFactor(config.defaultReplicationFactor.toShort).
           setDefaultNumPartitions(config.numPartitions.intValue()).
+          setReplicaPlacer(new PodReplicaPlacer(new StripedReplicaPlacer(new Random), config.canarySpec.toMap)).
           setSessionTimeoutNs(TimeUnit.NANOSECONDS.convert(config.brokerSessionTimeoutMs.longValue(),
             TimeUnit.MILLISECONDS)).
           setLeaderImbalanceCheckIntervalNs(leaderImbalanceCheckIntervalNs).

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -2537,6 +2537,7 @@ class KafkaApis(val requestChannel: RequestChannel,
             setHost(node.host).
             setPort(node.port).
             setRack(node.rack).
+            setPod(node.pod).
             setIsFenced(node.isFenced))
           }
         }

core/src/main/scala/kafka/server/KafkaConfig.scala

@@ -36,6 +36,7 @@ import org.apache.kafka.coordinator.group.Group.GroupType
 import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfig
 import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
 import org.apache.kafka.coordinator.share.ShareCoordinatorConfig
+import org.apache.kafka.metadata.placement.CanarySpec
 import org.apache.kafka.network.SocketServerConfigs
 import org.apache.kafka.raft.{KRaftConfigs, MetadataLogConfig, QuorumConfig}
 import org.apache.kafka.security.authorizer.AuthorizerUtils
@@ -163,6 +164,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
   val brokerSessionTimeoutMs: Int = getInt(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG)
   val controllerPerformanceSamplePeriodMs: Long = getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_SAMPLE_PERIOD_MS)
   val controllerPerformanceAlwaysLogThresholdMs: Long = getLong(KRaftConfigs.CONTROLLER_PERFORMANCE_ALWAYS_LOG_THRESHOLD_MS)
+  val canarySpec = new CanarySpec(getString(KRaftConfigs.CANARY_POD_NAME), getInt(KRaftConfigs.CANARY_PARTITION_INTERVAL))
 
   private def parseProcessRoles(): Set[ProcessRole] = {
     val roles = getList(KRaftConfigs.PROCESS_ROLES_CONFIG).asScala.map {

core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala

@@ -20,17 +20,19 @@ package kafka.server
 import java.util
 import java.util.{Collections, OptionalLong, Properties}
 import kafka.utils.TestUtils
+import org.apache.kafka.clients.MockClient.RequestMatcher
 import org.apache.kafka.common.Node
 import org.apache.kafka.common.Uuid
 import org.apache.kafka.common.message.{BrokerHeartbeatResponseData, BrokerRegistrationResponseData}
 import org.apache.kafka.common.protocol.Errors
 import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse, BrokerHeartbeatRequest, BrokerHeartbeatResponse, BrokerRegistrationRequest, BrokerRegistrationResponse}
 import org.apache.kafka.metadata.BrokerState
 import org.apache.kafka.raft.{KRaftConfigs, QuorumConfig}
-import org.apache.kafka.server.config.ServerLogConfigs
+import org.apache.kafka.server.config.{ServerLogConfigs, ServerConfigs}
 import org.apache.kafka.server.BrokerLifecycleManager
 import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api.{AfterEach, Test, Timeout}
+import org.junit.jupiter.api.{AfterEach, Assertions, Test, Timeout}
+
 
 import java.util.concurrent.{CompletableFuture, Future}
 import scala.jdk.CollectionConverters._
@@ -59,6 +61,7 @@ class BrokerLifecycleManagerTest {
     properties.setProperty(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "SSL")
     properties.setProperty(KRaftConfigs.INITIAL_BROKER_REGISTRATION_TIMEOUT_MS_CONFIG, "300000")
     properties.setProperty(KRaftConfigs.BROKER_HEARTBEAT_INTERVAL_MS_CONFIG, "100")
+    properties.setProperty(ServerConfigs.BROKER_POD_CONFIG, "pod1")
     properties
   }
 
@@ -97,7 +100,12 @@ class BrokerLifecycleManagerTest {
       assertEquals(1, context.mockChannelManager.unsentQueue.size)
       assertEquals(10L, context.mockChannelManager.unsentQueue.getFirst.request.build().asInstanceOf[BrokerRegistrationRequest].data().previousBrokerEpoch())
     }
-    context.mockClient.prepareResponseFrom(new BrokerRegistrationResponse(
+    val podMatches: RequestMatcher = { request =>
+      Assertions.assertNull(request.asInstanceOf[BrokerRegistrationRequest].data().rack())
+      Assertions.assertEquals("pod1", request.asInstanceOf[BrokerRegistrationRequest].data().pod())
+      true
+    }
+    context.mockClient.prepareResponseFrom(podMatches, new BrokerRegistrationResponse(
       new BrokerRegistrationResponseData().setBrokerEpoch(1000)), controllerNode)
     TestUtils.retry(10000) {
       context.poll()

core/src/test/scala/unit/kafka/server/DescribeClusterRequestTest.scala

@@ -62,8 +62,8 @@ class DescribeClusterRequestTest extends BaseRequestTest {
         .setBrokerId(server.config.brokerId)
         .setHost("localhost")
         .setPort(server.socketServer.boundPort(listenerName))
-        .setRack(server.config.rack.orElse(null))
-    }.toSet
+        .setRack(server.config.rack().orElse(null))
+        .setPod(server.config.pod().orElse(null))}.toSet
 
     val expectedClusterId = brokers.last.clusterId
 

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

@@ -4418,9 +4418,10 @@ class KafkaApisTest extends Logging {
         .setSecurityProtocol(SecurityProtocol.PLAINTEXT.id)
         .setName(plaintextListener.value)
     )
+
     MetadataCacheFixtures.updateCache(metadataCache,
-      util.List.of(new RegisterBrokerRecord().setBrokerId(0).setRack("rack").setFenced(false).setEndPoints(endpoints))
-    )
+      util.List.of(new RegisterBrokerRecord().setBrokerId(0).setRack("rack").setPod("pod").setFenced(false).setEndPoints(endpoints))
+)
 
     // 2. Set up authorizer
     val authorizer: Authorizer = mock(classOf[Authorizer])
@@ -10177,6 +10178,7 @@ class KafkaApisTest extends Logging {
       util.List.of(new RegisterBrokerRecord()
         .setBrokerId(brokerId)
         .setRack("rack")
+        .setPod("pod")
         .setFenced(false)
         .setEndPoints(endpoints)))
 
@@ -10201,7 +10203,6 @@ class KafkaApisTest extends Logging {
   private def updateMetadataCacheWithInconsistentListeners(): (ListenerName, ListenerName) = {
     val plaintextListener = ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT)
     val anotherListener = new ListenerName("LISTENER2")
-
     val endpoints0 = new BrokerEndpointCollection()
     endpoints0.add(
       new BrokerEndpoint()
@@ -10229,8 +10230,7 @@ class KafkaApisTest extends Logging {
 
     MetadataCacheFixtures.updateCache(metadataCache,
       util.List.of(new RegisterBrokerRecord().setBrokerId(0).setRack("rack").setFenced(false).setEndPoints(endpoints0),
-      new RegisterBrokerRecord().setBrokerId(1).setRack("rack").setFenced(false).setEndPoints(endpoints1))
-    )
+      new RegisterBrokerRecord().setBrokerId(1).setRack("rack").setPod("pod").setFenced(false).setEndPoints(endpoints1)))
 
     (plaintextListener, anotherListener)
   }
@@ -10483,6 +10483,7 @@ class KafkaApisTest extends Logging {
     new RegisterBrokerRecord()
       .setBrokerId(brokerId)
       .setRack("rack")
+      .setPod("pod")
       .setFenced(false)
       .setEndPoints(endpoints)
       .setBrokerEpoch(brokerEpoch)

core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala

@@ -19,7 +19,7 @@ package kafka.server
 
 import java.net.InetSocketAddress
 import java.util
-import java.util.{Arrays, Collections, Properties}
+import java.util.{Arrays, Collections, Optional, Properties}
 import kafka.utils.TestUtils.assertBadConfigContainingMessage
 import kafka.utils.TestUtils
 import org.apache.kafka.common.{Endpoint, Node}
@@ -823,6 +823,8 @@ class KafkaConfigTest {
         case MetadataLogConfig.INTERNAL_METADATA_MAX_BATCH_SIZE_IN_BYTES_CONFIG => // no op
         case MetadataLogConfig.INTERNAL_METADATA_DELETE_DELAY_MILLIS_CONFIG => // no op
         case KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG => // ignore string
+        case KRaftConfigs.CANARY_POD_NAME => assertPropertyInvalid(baseProperties, name, " ")
+        case KRaftConfigs.CANARY_PARTITION_INTERVAL => assertPropertyInvalid(baseProperties, name, "not_a_number", "-1")
         case MetadataLogConfig.METADATA_MAX_IDLE_INTERVAL_MS_CONFIG  => assertPropertyInvalid(baseProperties, name, "not_a_number")
 
         case ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG => //ignore string
@@ -919,6 +921,7 @@ class KafkaConfigTest {
         case MetricConfigs.METRIC_REPORTER_CLASSES_CONFIG => // ignore string
         case MetricConfigs.METRIC_RECORDING_LEVEL_CONFIG => // ignore string
         case ServerConfigs.BROKER_RACK_CONFIG => // ignore string
+        case ServerConfigs.BROKER_POD_CONFIG => // ignore string
 
         case ServerConfigs.COMPRESSION_GZIP_LEVEL_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", "0")
         case ServerConfigs.COMPRESSION_LZ4_LEVEL_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", "0")
@@ -2031,4 +2034,19 @@ class KafkaConfigTest {
         "Please increase broker.session.timeout.ms or decrease broker.heartbeat.interval.ms."))
     }
   }
+
+  @Test
+  def testPodAndRackProperties(): Unit = {
+    val props = new Properties()
+    props.put(KRaftConfigs.PROCESS_ROLES_CONFIG, "broker")
+    props.setProperty(QuorumConfig.QUORUM_VOTERS_CONFIG, "2@localhost:9093")
+    props.setProperty(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "CONTROLLER")
+    props.put(KRaftConfigs.NODE_ID_CONFIG, "1")
+    props.put(ServerConfigs.BROKER_RACK_CONFIG, "rack-1")
+    props.put(ServerConfigs.BROKER_POD_CONFIG, "pod-1")
+    assertTrue(isValidKafkaConfig(props))
+    val config = KafkaConfig.fromProps(props)
+    assertEquals(Optional.of("rack-1"), config.rack)
+    assertEquals(Optional.of("pod-1"), config.pod)
+  }
 }