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

[ericzhifengchen]

Implements the broker-side placement portion of KIP-1095. Adds an optional
broker attribute, pod, and a replica placement layer that can confine a
configurable subset of partitions to a single pod.

https://cwiki.apache.org/confluence/display/KAFKA/KIP-1095%3A+Kafka+Canary+Isolation

Motivation

A bad broker deployment can affect any partition hosted on the brokers being
upgraded, and operators have no way to bound that set in advance. Confining a
known fraction of partitions to a designated pod makes the blast radius of a
regression a deployment-time decision rather than an after-the-fact
discovery.

Rack-awareness cannot express this: rack means "spread replicas across these
groups", whereas this requires "confine these partitions to this group".

Changes

Broker pods. New broker config broker.pod (string, default null),
exposed as AbstractKafkaConfig.pod(). The value travels through broker
registration into the metadata log and out through Metadata and
DescribeCluster, where it becomes Node.pod(). A Pod field is added as
a tagged field to four messages, so no message version is bumped:

Message	Field versions	Tag
BrokerRegistrationRequest	0+	0
RegisterBrokerRecord	0+	2
DescribeClusterResponse	0+	0
MetadataResponse	11+	0

Pod isolation. PodReplicaPlacer decorates the configured
ReplicaPlacer with a map of pod name to a predicate over partition index.
A partition matching exactly one rule is placed only on brokers in that pod;
a partition matching none is placed across the pods carrying no rule.
Consecutive partitions resolving to the same broker set are coalesced into a
single PlacementSpec, so the delegate still sees contiguous ranges and its
striping and rack-awareness are unchanged.

A broker with no pod belongs to every pod, which allows incremental
adoption. A rule naming a pod with no live brokers is ignored and falls
through to default placement. A partition matching more than one rule raises
IllegalStateException naming the conflicting pods. A pod smaller than the
requested replication factor fails with InvalidReplicationFactorException
rather than spilling replicas outside the pod.

Canary isolation. Two controller configs install the single rule this
KIP configures: canary.pod.name (default canary-broker) and
canary.partition.interval (int, default 0). With interval N, partition i
is a canary partition when i % N == N - 1, so a topic with fewer than N
partitions has none. 0 disables the feature.

Tooling. kafka-reassign-partitions --generate accepts --canary-name
and --canary-interval so that offline plans match controller placement.
Without them a reassignment would silently undo isolation on the topics it
moves. This also fixes broker pod being dropped from generated plans when
rack-awareness was disabled.

Compatibility

Off by default. With canary.partition.interval at 0 the rule map is empty
and placement is byte-identical to StripedReplicaPlacer.

All protocol additions are tagged, nullable and default to null, so no
message version is bumped and no API version negotiation changes. Older
clients skip the unknown tag. A broker that does not set broker.pod
behaves exactly as it does today.

Note that isolation is only complete once every broker declares a pod, since
an unlabelled broker is eligible for every pod. A partially labelled cluster
is a migration state, not a supported configuration.

Testing

• PodReplicaPlacerTest (13 cases): placement with zero, one and multiple
  pods; pod-less brokers; empty clusters; partition addition to existing
  topics; and the overlapping-rule error.
• CanarySpecTest (8 cases): interval-to-predicate conversion, including 0,
  negative, interval 1, and the property that a topic smaller than the
  interval has no canary partition.
• BrokerRegistrationTest, MetadataCacheTest, ClusterImageBrokersNodeTest
  and BrokerLifecycleManagerTest: propagation through registration, the
  metadata log, the metadata cache and Node.
• KafkaConfigTest: the new broker config and validation of the new
  controller config.
• ReassignPartitionsUnitTest: generated plans with and without canary
  placement.