// Path: src/main/java/kafka/utils/Function2.java
// public interface Function2<A1,A2,B> {
//     B apply(A1 arg1, A2 arg2);
// }
// 
// Path: src/main/java/kafka/utils/SystemTime.java
// public class SystemTime implements Time {
//     public static final SystemTime instance = new SystemTime();
// 
//     @Override
//     public long milliseconds() {
//         return System.currentTimeMillis();
//     }
// 
//     @Override
//     public long nanoseconds() {
//         return System.nanoTime();
//     }
// 
//     @Override
//     public void sleep(long ms) {
//         try {
//             Thread.sleep(ms);
//         } catch (InterruptedException e) {
// 
//         }
//     }
// }


import com.alibaba.fastjson.JSON;
import com.google.common.collect.ImmutableMap;
import kafka.controller.ControllerContext;
import kafka.controller.KafkaControllers;
import kafka.utils.Callable0;
import kafka.utils.Function2;
import kafka.utils.SystemTime;
import kafka.utils.ZkUtils;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static kafka.utils.ZkUtils.*;
        this.onBecomingLeader = onBecomingLeader;
        this.brokerId = brokerId;

        index = electionPath.lastIndexOf("/");
        if (index > 0)
            makeSurePersistentPathExists(controllerContext.zkClient, electionPath.substring(0, index));
    }

    public int leaderId = -1;
    // create the election path in ZK, if one does not exist
    public int index;

    public LeaderChangeListener leaderChangeListener = new LeaderChangeListener();
    Logger logger = LoggerFactory.getLogger(ZookeeperLeaderElector.class);

    @Override
    public void startup() {
        synchronized (controllerContext.controllerLock) {
            controllerContext.zkClient.subscribeDataChanges(electionPath, leaderChangeListener);
            elect();
        }
    }

    @Override
    public boolean amILeader() {
        return leaderId == brokerId;
    }

    @Override
    public boolean elect() {
// Path: src/main/java/kafka/utils/Function2.java
// public interface Function2<A1,A2,B> {
//     B apply(A1 arg1, A2 arg2);
// }
// 
// Path: src/main/java/kafka/utils/SystemTime.java
// public class SystemTime implements Time {
//     public static final SystemTime instance = new SystemTime();
// 
//     @Override
//     public long milliseconds() {
//         return System.currentTimeMillis();
//     }
// 
//     @Override
//     public long nanoseconds() {
//         return System.nanoTime();
//     }
// 
//     @Override
//     public void sleep(long ms) {
//         try {
//             Thread.sleep(ms);
//         } catch (InterruptedException e) {
// 
//         }
//     }
// }


// Path: src/main/java/kafka/server/ZookeeperLeaderElector.java
import com.alibaba.fastjson.JSON;
import com.google.common.collect.ImmutableMap;
import kafka.controller.ControllerContext;
import kafka.controller.KafkaControllers;
import kafka.utils.Callable0;
import kafka.utils.Function2;
import kafka.utils.SystemTime;
import kafka.utils.ZkUtils;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static kafka.utils.ZkUtils.*;

        this.onBecomingLeader = onBecomingLeader;
        this.brokerId = brokerId;

        index = electionPath.lastIndexOf("/");
        if (index > 0)
            makeSurePersistentPathExists(controllerContext.zkClient, electionPath.substring(0, index));
    }

    public int leaderId = -1;
    // create the election path in ZK, if one does not exist
    public int index;

    public LeaderChangeListener leaderChangeListener = new LeaderChangeListener();
    Logger logger = LoggerFactory.getLogger(ZookeeperLeaderElector.class);

    @Override
    public void startup() {
        synchronized (controllerContext.controllerLock) {
            controllerContext.zkClient.subscribeDataChanges(electionPath, leaderChangeListener);
            elect();
        }
    }

    @Override
    public boolean amILeader() {
        return leaderId == brokerId;
    }

    @Override
    public boolean elect() {
        String timestamp = SystemTime.instance.milliseconds() + "";