...

However, there are still special cases—such as internal testing or experimental features—where users may want to define a custom placement strategy. Therefore, we propose providing a mechanism that allows users to implement their own ReplicaPlacer.

Previously, KIP-660: Pluggable ReplicaPlacer proposed  proposed making the ReplicaPlacer pluggable, but the proposal was rejected. The main concern was that an intelligent placement strategy might require a full view of the cluster, which could be prohibitively expensive in large deployments. Furthermore, such implementations could become overly complex and might introduce latency that impacts the controller thread, thus affecting system stability.

...

• Class: ReplicationConfigs
• Name: replica.placer.class.name
• Type: string
• Default: StripedReplicaPlacer
• Valid Values: non-empty string
• DocDocument: The fully qualified class name of the ReplicaPlacer implementation. This class is used by the broker to determine the placement of replicas when topics or partitions are created. This configuration is not intended for external users and should only be used in the following scenarios: internal testing environments, experimental feature development, performance evaluation or benchmarking. Custom implementations must be lightweight and efficient to avoid negatively impacting controller performance and overall cluster stability.

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 class is used by the broker to determine the placement of replicas when topics or partitions are created. " +
            "This configuration is not intended for external users and should only be used in the following scenarios: " +
            "internal testing environments, experimental feature development, performance evaluation or benchmarking. \n " +
            "Custom implementations must be lightweight and efficient to avoid negatively impacting controller performance " +
            "and overall cluster stability.";

Update the ReplicaPlacer Interface, which need The ReplicaPlacer interface has been updated to extend both Configurable , Closeableand Closeable to support proper configuration and resource cleanup.

package org.apache.kafka.metadata.placement;

import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.errors.InvalidReplicationFactorException;


/**
 * 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;
}

...

Add new test for new internal config internal.replica.placer.class.name

Rejected Alternatives

KIP-660: Pluggable ReplicaPlacer, but the proposal was ultimately rejected due to the following concerns:

...