// 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.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Predicate;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import joptsimple.OptionParser;
import joptsimple.OptionSet;
import joptsimple.OptionSpec;
import kafka.common.AdminCommandFailedException;
import kafka.common.TopicAndPartition;
import kafka.utils.*;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
package kafka.admin;



public class PreferredReplicaLeaderElectionCommand {
    public static void main(String[] args) {
        OptionParser parser = new OptionParser();
        OptionSpec<String> jsonFileOpt = parser.accepts("path-to-json-file", "The JSON file with the list of partitions " +
                "for which preferred replica leader election should be done, in the following format - \n" +
                "{\"partitions\":\n\t[{\"topic\": \"foo\", \"partition\": 1},\n\t {\"topic\": \"foobar\", \"partition\": 2}]\n}\n" +
                "Defaults to all existing partitions")
                .withRequiredArg()
                .describedAs("list of partitions for which preferred replica leader election needs to be triggered")
                .ofType(String.class);
        OptionSpec<String> zkConnectOpt = parser.accepts("zookeeper", "REQUIRED: The connection string for the zookeeper connection in the " +
                "form host:port. Multiple URLS can be given to allow fail-over.")
                .withRequiredArg()
                .describedAs("urls")
                .ofType(String.class);

        OptionSet options = parser.parse(args);

        CommandLineUtils.checkRequiredArgs(parser, options, zkConnectOpt);

        String zkConnect = options.valueOf(zkConnectOpt);
        ZkClient zkClient = null;

        try {
            zkClient = new ZkClient(zkConnect, 30000, 30000, ZKStringSerializer.instance);
// 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/admin/PreferredReplicaLeaderElectionCommand.java
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Predicate;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import joptsimple.OptionParser;
import joptsimple.OptionSet;
import joptsimple.OptionSpec;
import kafka.common.AdminCommandFailedException;
import kafka.common.TopicAndPartition;
import kafka.utils.*;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;

package kafka.admin;



public class PreferredReplicaLeaderElectionCommand {
    public static void main(String[] args) {
        OptionParser parser = new OptionParser();
        OptionSpec<String> jsonFileOpt = parser.accepts("path-to-json-file", "The JSON file with the list of partitions " +
                "for which preferred replica leader election should be done, in the following format - \n" +
                "{\"partitions\":\n\t[{\"topic\": \"foo\", \"partition\": 1},\n\t {\"topic\": \"foobar\", \"partition\": 2}]\n}\n" +
                "Defaults to all existing partitions")
                .withRequiredArg()
                .describedAs("list of partitions for which preferred replica leader election needs to be triggered")
                .ofType(String.class);
        OptionSpec<String> zkConnectOpt = parser.accepts("zookeeper", "REQUIRED: The connection string for the zookeeper connection in the " +
                "form host:port. Multiple URLS can be given to allow fail-over.")
                .withRequiredArg()
                .describedAs("urls")
                .ofType(String.class);

        OptionSet options = parser.parse(args);

        CommandLineUtils.checkRequiredArgs(parser, options, zkConnectOpt);

        String zkConnect = options.valueOf(zkConnectOpt);
        ZkClient zkClient = null;

        try {
            zkClient = new ZkClient(zkConnect, 30000, 30000, ZKStringSerializer.instance);
            Set<TopicAndPartition> partitionsForPreferredReplicaElection =