Page History

/**
 * This interface provides an API for implementing replica assignment strategies. Both rack aware and unaware
 * strategies are supported by this interface. If needed it can defer assignment to Kafka's default rack unaware
 * replicas assignment strategy.
 */
public interface BrokerReplicaAssignor {

    /**
     * Creates an assignment of a topic given the racks, number of partitions and the replication factor.
     * @param clusterView is the necessary information about the cluster (such as current assignment, rack information)
     *                    that is needed to be able to decide replica assignments.
     * @param numPartitions is the number of partitions of the assigned topic.
     * @param replicationFactor is the replication factor of the assigned topic.
     * @return
@param brokerIds is an optional *list <ul>
of brokerIDs that is used *to generate an assignment. If <li><code>null</code>null tois deferspecified,
 the assignment to Kafka, that* uses its rack unaware assignment strategy.</li>
     *     <li>the assignment inthen theall formbrokers of (partition) -> (array of broker IDs) mapping where the size of the mapare used.
     * @return
     * <ul>
     *     equals<li><code>null</code> to the number of partitions anddefer the sizeassignment ofto valuesKafka, inthat theuses mapits equalsrack tounaware theassignment replication factorstrategy.</li>
     * </ul>
    <li>the *assignment @throws org.apache.kafka.common.errors.InvalidReplicaAssignmentException if the algorithm can't assign replicas
     * based on the specified parameters and can't defer it to Kafka's default rack unaware assignment strategy.
     * in the form of (partition) -> (array of broker IDs) mapping where the size of the map
     *     equals to the number of partitions and the size of values in the map equals to the replication factor.</li>
     * </ul>
     * @throws org.apache.kafka.common.errors.InvalidReplicationFactorExceptionInvalidReplicaAssignmentException if the algorithm numbercan't ofassign replicas is greater
     * thanbased whaton the assignmentspecified strategy accepts as a maximum valueparameters and can't defer it to Kafka's default rack unaware assignment strategy.
     */
    Map<Integer, Integer[]> assignReplicasToTopic(ClusterView clusterView, int numPartitions, short replicationFactor);
}

/**
 * This class defines a view of a cluster that the partition assignor can see, such as the current assignment
 * or broker -> rack mappings.
 */
public class ClusterView {

    private final Map<TopicPartition, List<Integer>> partitionAssignments;
    private final List<BrokerInfo> brokerInfos;

    public ClusterView(Map<TopicPartition, List<Integer>> partitionAssignments, List<BrokerInfo> brokerInfos) {
 @throws org.apache.kafka.common.errors.InvalidReplicationFactorException if the number of replicas is greater
     * than what the assignment strategy accepts as a maximum value.
     */
    Map<Integer, Integer[]> assignReplicasToTopic(ClusterView clusterView, int numPartitions, short replicationFactor,
                                     this.partitionAssignments = partitionAssignments;
        this.brokerInfos = brokerInfos;
 List<Integer>   }

    public Map<TopicPartition, List<Integer>> topicAssignments() {
        return Collections.unmodifiableMap(partitionAssignments);
    }

    public List<BrokerInfo> brokerInfos() {
        return Collections.unmodifiableList(brokerInfos);
    }
}brokerIds);
}  

public class CreateReplicaAssignmentOptions extends CreateReplicaAssignmentResultAbstractOptions<CreateReplicaAssignmentOptions> {

    private KafkaFuture<Map<Integer, Integer[]>> futureboolean forceSimpleAssignment;
    private List<Integer> brokerIds;

    CreateReplicaAssignmentResult(KafkaFuture<Map<Integer, Integer[]>> futurepublic boolean forceSimpleAssignment() {
        this.future = futurereturn forceSimpleAssignment;
    }

    public KafkaFuture<Map<Integer, Integer[]>> assignments(CreateReplicaAssignmentOptions forceSimpleAssignment(boolean forceSimpleAssignment) {
        return futurethis.forceSimpleAssignment = forceSimpleAssignment;
    }
}

public class CreateReplicaAssignmentOptions extends AbstractOptions<CreateReplicaAssignmentOptions>return {this;

    private boolean forceSimpleAssignment;}

    public booleanList<Integer> forceSimpleAssignmentbrokerIds() {
        return forceSimpleAssignmentCollections.unmodifiableList(brokerIds);
    }

    public CreateReplicaAssignmentOptions forceSimpleAssignmentbrokerIds(booleanList<Integer> forceSimpleAssignmentbrokerIds) {
        this.forceSimpleAssignmentbrokerIds = forceSimpleAssignmentbrokerIds;
        return this;
    }
}