// Path: src/main/java/kafka/common/TopicAndPartition.java
// public class TopicAndPartition {
//     public final String topic;
//     public final int partition;
// 
//     public TopicAndPartition(String topic, int partition) {
//         this.topic = topic;
//         this.partition = partition;
//     }
// 
//     public TopicAndPartition(Tuple2<String, Integer> tuple) {
//         this(tuple._1, tuple._2);
//     }
// 
//     public TopicAndPartition(Partition partition) {
//         this(partition.topic, partition.partitionId);
//     }
// 
//     public TopicAndPartition(Replica replica) {
//         this(replica.topic, replica.partitionId);
//     }
// 
//     public Tuple2<String, Integer> asTuple() {
//         return Tuple2.make(topic, partition);
//     }
// 
//     @Override
//     public String toString() {
//         return String.format("[%s,%d]", topic, partition);
//     }
// 
//     @Override
//     public boolean equals(Object o) {
//         if (this == o) return true;
//         if (o == null || getClass() != o.getClass()) return false;
// 
//         TopicAndPartition that = (TopicAndPartition) o;
// 
//         if (partition != that.partition) return false;
//         if (topic != null ? !topic.equals(that.topic) : that.topic != null) return false;
// 
//         return true;
//     }
// 
//     @Override
//     public int hashCode() {
//         int result = topic != null ? topic.hashCode() : 0;
//         result = 31 * result + partition;
//         return result;
//     }
// }


import com.google.common.base.Predicate;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import kafka.api.RequestOrResponse;
import kafka.cluster.Broker;
import kafka.common.KafkaException;
import kafka.common.StateChangeFailedException;
import kafka.common.TopicAndPartition;
import kafka.utils.*;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
     * @param replicas    The list of replicas (brokers) that need to be transitioned to the target state
     * @param targetState The state that the replicas should be moved to
     *                    The controller's allLeaders cache should have been updated before this
     */
    public void handleStateChanges(Set<PartitionAndReplica> replicas, final ReplicaState targetState) {
        logger.info("Invoking state change to {} for replicas {}", targetState, replicas);
        try {
            brokerRequestBatch.newBatch();
            Utils.foreach(replicas, new Callable1<PartitionAndReplica>() {
                @Override
                public void apply(PartitionAndReplica r) {
                    handleStateChange(r.topic, r.partition, r.replica, targetState);
                }
            });
            brokerRequestBatch.sendRequestsToBrokers(controller.epoch(), controllerContext.correlationId.getAndIncrement());
        } catch (Throwable e) {
            logger.error("Error while moving some replicas to {} state", targetState, e);
        }
    }

    /**
     * This API exercises the replica's state machine. It ensures that every state transition happens from a legal
     * previous state to the target state.
     *
     * @param topic       The topic of the replica for which the state transition is invoked
     * @param partition   The partition of the replica for which the state transition is invoked
     * @param replicaId   The replica for which the state transition is invoked
     * @param targetState The end state that the replica should be moved to
     */
    public void handleStateChange(String topic, int partition, final int replicaId, ReplicaState targetState) {
// Path: src/main/java/kafka/common/TopicAndPartition.java
// public class TopicAndPartition {
//     public final String topic;
//     public final int partition;
// 
//     public TopicAndPartition(String topic, int partition) {
//         this.topic = topic;
//         this.partition = partition;
//     }
// 
//     public TopicAndPartition(Tuple2<String, Integer> tuple) {
//         this(tuple._1, tuple._2);
//     }
// 
//     public TopicAndPartition(Partition partition) {
//         this(partition.topic, partition.partitionId);
//     }
// 
//     public TopicAndPartition(Replica replica) {
//         this(replica.topic, replica.partitionId);
//     }
// 
//     public Tuple2<String, Integer> asTuple() {
//         return Tuple2.make(topic, partition);
//     }
// 
//     @Override
//     public String toString() {
//         return String.format("[%s,%d]", topic, partition);
//     }
// 
//     @Override
//     public boolean equals(Object o) {
//         if (this == o) return true;
//         if (o == null || getClass() != o.getClass()) return false;
// 
//         TopicAndPartition that = (TopicAndPartition) o;
// 
//         if (partition != that.partition) return false;
//         if (topic != null ? !topic.equals(that.topic) : that.topic != null) return false;
// 
//         return true;
//     }
// 
//     @Override
//     public int hashCode() {
//         int result = topic != null ? topic.hashCode() : 0;
//         result = 31 * result + partition;
//         return result;
//     }
// }


// Path: src/main/java/kafka/controller/ReplicaStateMachine.java
import com.google.common.base.Predicate;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import kafka.api.RequestOrResponse;
import kafka.cluster.Broker;
import kafka.common.KafkaException;
import kafka.common.StateChangeFailedException;
import kafka.common.TopicAndPartition;
import kafka.utils.*;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

     * @param replicas    The list of replicas (brokers) that need to be transitioned to the target state
     * @param targetState The state that the replicas should be moved to
     *                    The controller's allLeaders cache should have been updated before this
     */
    public void handleStateChanges(Set<PartitionAndReplica> replicas, final ReplicaState targetState) {
        logger.info("Invoking state change to {} for replicas {}", targetState, replicas);
        try {
            brokerRequestBatch.newBatch();
            Utils.foreach(replicas, new Callable1<PartitionAndReplica>() {
                @Override
                public void apply(PartitionAndReplica r) {
                    handleStateChange(r.topic, r.partition, r.replica, targetState);
                }
            });
            brokerRequestBatch.sendRequestsToBrokers(controller.epoch(), controllerContext.correlationId.getAndIncrement());
        } catch (Throwable e) {
            logger.error("Error while moving some replicas to {} state", targetState, e);
        }
    }

    /**
     * This API exercises the replica's state machine. It ensures that every state transition happens from a legal
     * previous state to the target state.
     *
     * @param topic       The topic of the replica for which the state transition is invoked
     * @param partition   The partition of the replica for which the state transition is invoked
     * @param replicaId   The replica for which the state transition is invoked
     * @param targetState The end state that the replica should be moved to
     */
    public void handleStateChange(String topic, int partition, final int replicaId, ReplicaState targetState) {
        TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);