...

The issue of KAFKA-19507 shows a real-world need for solving custom replica assignment problems. It reinforces the point that different usage scenarios often come with unique assignment needs.

Nevertheless, we can make the classify ReplicaPlacer as an internal pluggable unstable component. This means the mechanism would not will be exposed to external users, but could be utilized without any guarantees of compatibility or long-term stability. The configuration is primarily intended for advanced users and can be useful in internal testing, special-purpose specialized scenarios, or controlled experiments. By allowing different implementations to be injected internally, we can make gain flexibility in the placement logic more flexible for testing purposes, without compromising while ensuring that the stability of the controller is not compromised.

Public Interfaces

New internal broker configuration:

• Class: ReplicationConfigs
• Name: replica.placer.class.name
• Type: string
• Default: StripedReplicaPlacer
• Valid Values: non-empty string
• Document: The fully qualified class name of the ReplicaPlacer implementation. This class plugin is considered unstable, meaning the Kafka community does not provide guarantees of compatibility or long-term stability. It is used by the broker to determine the replica placement of replicas when topics or partitions are created. This configuration is not intended for external users and should be used in the following scenarios: advanced use cases such as internal testing environments, experimental feature development, performance evaluation, or benchmarking. Custom implementations must should be lightweight and efficient to avoid negatively impacting degrading controller performance and or impacting overall cluster stability.

New public package:

• org.apache.kafka.metadata.placement

New public unstable classes in `org.apache.kafka.metadata.placement` package

• ReplicaPlacer
• PartitionAssignment

• TopicAssignment

• PlacementSpec

• DefaultDirProvider

• ClusterDescriber

• UsableBroker

Proposed Changes

Add a new internal configuration: internal. replica.placer.class.name. This configuration allows users to define a custom ReplicaPlacer implementation.

    public static final String REPLICA_PLACER_CLASS_NAME = "replica.placer.class.name";
    public static final String REPLICA_PLACER_CLASS_NAME_DEFAULT = StripedReplicaPlacer.class.getName();     
	public static final String REPLICA_PLACER_CLASS_NAME_DOC = "The fully qualified class name of the " +
        "ReplicaPlacer implementation. This plugin is considered unstable, meaning the Kafka " +
        "community does not provide  "This classguarantees of compatibility or long-term stability. It is used " +
        "by the broker to determine thereplica placement of replicas when topics or partitions are created. This " +
            "This configuration is not intended for externaladvanced usersuse andcases shouldsuch beas usedinternal in the following scenarios:testing environments, " +
            "internal testing environments, experimental feature development, performance evaluation, or benchmarking. \nCustom " +
            "Custom implementations mustshould be lightweight and efficient to avoid negatively impactingdegrading controller performance " +
        " or   "andimpacting overall cluster stability.";

...

package org.apache.kafka.metadata.placement;

/**
 * The interface which a Kafka replica placement policy must implement.
 */
@InterfaceStability.Unstable
public interface ReplicaPlacer extends Configurable, Closeable {
    /**
     * Create a new replica placement.
     *
     * @param placement     What we're trying to place.
     * @param cluster       A description of the cluster we're trying to place in.
     *
     * @return              A topic assignment.
     *
     * @throws InvalidReplicationFactorException    If too many replicas were requested.
     */
    TopicAssignment place(
        PlacementSpec placement,
        ClusterDescriber cluster
    ) throws InvalidReplicationFactorException;
}

The following classes should also be marked with @InterfaceStability.Unstable.

• ReplicaPlacer
• PartitionAssignment

• TopicAssignment

• PlacementSpec

• DefaultDirProvider

• ClusterDescriber

• UsableBroker

Also move StripedReplicaPlacer into internal package `org.apache.kafka.metadata.placement.internal`, and add `package-info.java` into `org.apache.kafka.metadata.placement`

/**
 * Provides Replica placer and related classes. 
 */
package org.apache.kafka.metadata.placement;

Compatibility, Deprecation, and Migration Plan

This KIP adds internal new config which doesn’t break compatibility.

...

• Implementing an intelligent placement strategy may require access to the full cluster state, which could be prohibitively expensive in large deployments.

• Such strategies could become overly complex and harder to maintain.

• Slow or inefficient placement logic could block the controller thread, negatively impacting Kafka's overall stability.

Make replica.placer.class.name as an interanl config

• The advantage is that we are not required to maintain compatibility since it is not exposed to users.
• The downside is that most users would not be aware of the configuration at all.
• Making this configuration private will provide less opportunities to gather feedback from users.

Using Admin API for Post-Creation Reassignment

...