// Path: src/main/java/com/conductor/kafka/Broker.java
// public class Broker implements Writable {
// 
//     private int id;
//     private String host;
//     private int port;
// 
//     /**
//      * The {@link Writable} constructor; use {@link #Broker(String, int, int)}
//      */
//     public Broker() {
//     }
// 
//     public Broker(final String host, final int port, final int id) {
//         this.port = port;
//         this.host = host;
//         this.id = id;
//     }
// 
//     public int getId() {
//         return id;
//     }
// 
//     public String getHost() {
//         return host;
//     }
// 
//     public int getPort() {
//         return port;
//     }
// 
//     public void setId(final int id) {
//         this.id = id;
//     }
// 
//     public void setHost(final String host) {
//         this.host = host;
//     }
// 
//     public void setPort(final int port) {
//         this.port = port;
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void write(final DataOutput dataOutput) throws IOException {
//         dataOutput.writeUTF(host);
//         dataOutput.writeInt(port);
//         dataOutput.writeInt(id);
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void readFields(final DataInput dataInput) throws IOException {
//         this.host = dataInput.readUTF();
//         this.port = dataInput.readInt();
//         this.id = dataInput.readInt();
//     }
// 
//     @Override
//     public String toString() {
//         return Objects.toStringHelper(this).add("id", id).add("host", host).add("port", port).toString();
//     }
// 
//     @Override
//     public boolean equals(final Object o) {
//         if (this == o)
//             return true;
//         if (!(o instanceof Broker))
//             return false;
// 
//         final Broker broker = (Broker) o;
// 
//         if (id != broker.id)
//             return false;
//         if (port != broker.port)
//             return false;
//         if (host != null ? !host.equals(broker.host) : broker.host != null)
//             return false;
// 
//         return true;
//     }
// 
//     @Override
//     public int hashCode() {
//         int result = id;
//         result = 31 * result + (host != null ? host.hashCode() : 0);
//         result = 31 * result + port;
//         return result;
//     }
// }
// 
// Path: src/main/java/com/conductor/kafka/zk/ZkUtils.java
// @VisibleForTesting
// static class StringSerializer implements ZkSerializer {
// 
//     public StringSerializer() {
//     }
// 
//     @Override
//     public Object deserialize(byte[] data) throws ZkMarshallingError {
//         if (data == null)
//             return null;
//         return new String(data);
//     }
// 
//     @Override
//     public byte[] serialize(Object data) throws ZkMarshallingError {
//         return data.toString().getBytes();
//     }
// 
// }


import com.conductor.kafka.Broker;
import com.conductor.kafka.Partition;
import com.conductor.kafka.zk.ZkUtils.StringSerializer;
import com.google.common.collect.Lists;
import org.I0Itec.zkclient.ZkClient;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import java.util.List;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;
        // with default namespace
        assertEquals("", zk.getZkRoot());

        // non-default namespace, tailing "/"
        zk = new ZkUtils(client, "/com/conductor/");
        assertEquals("/com/conductor", zk.getZkRoot());
    }

    @Test
    public void testGetBrokerId() throws Exception {
        // normal case
        final int brokerId = 1;
        when(client.readData("/brokers/ids/" + brokerId, true)).thenReturn("{ \"host\":\"127.0.0.1\", \"jmx_port\":9999, \"port\":9092, \"version\":1 }");
        final Broker broker = zk.getBroker(brokerId);
        assertEquals("127.0.0.1", broker.getHost());
        assertEquals(9092, broker.getPort());
        assertEquals(brokerId, broker.getId());

        // non-existant broker
        when(client.readData("/brokers/ids/" + 2, true)).thenReturn(null);
        final Broker nonExistantBroker = zk.getBroker(2);
        assertNull(nonExistantBroker);
    }

    @Test
    public void testGetBrokers() throws Exception {
        final List<String> brokerIds = Lists.newArrayList("1", "2");
        final Broker b1 = new Broker("localhost", 9092, 1);
        final Broker b2 = new Broker("localhost", 9092, 2);

// Path: src/main/java/com/conductor/kafka/Broker.java
// public class Broker implements Writable {
// 
//     private int id;
//     private String host;
//     private int port;
// 
//     /**
//      * The {@link Writable} constructor; use {@link #Broker(String, int, int)}
//      */
//     public Broker() {
//     }
// 
//     public Broker(final String host, final int port, final int id) {
//         this.port = port;
//         this.host = host;
//         this.id = id;
//     }
// 
//     public int getId() {
//         return id;
//     }
// 
//     public String getHost() {
//         return host;
//     }
// 
//     public int getPort() {
//         return port;
//     }
// 
//     public void setId(final int id) {
//         this.id = id;
//     }
// 
//     public void setHost(final String host) {
//         this.host = host;
//     }
// 
//     public void setPort(final int port) {
//         this.port = port;
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void write(final DataOutput dataOutput) throws IOException {
//         dataOutput.writeUTF(host);
//         dataOutput.writeInt(port);
//         dataOutput.writeInt(id);
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void readFields(final DataInput dataInput) throws IOException {
//         this.host = dataInput.readUTF();
//         this.port = dataInput.readInt();
//         this.id = dataInput.readInt();
//     }
// 
//     @Override
//     public String toString() {
//         return Objects.toStringHelper(this).add("id", id).add("host", host).add("port", port).toString();
//     }
// 
//     @Override
//     public boolean equals(final Object o) {
//         if (this == o)
//             return true;
//         if (!(o instanceof Broker))
//             return false;
// 
//         final Broker broker = (Broker) o;
// 
//         if (id != broker.id)
//             return false;
//         if (port != broker.port)
//             return false;
//         if (host != null ? !host.equals(broker.host) : broker.host != null)
//             return false;
// 
//         return true;
//     }
// 
//     @Override
//     public int hashCode() {
//         int result = id;
//         result = 31 * result + (host != null ? host.hashCode() : 0);
//         result = 31 * result + port;
//         return result;
//     }
// }
// 
// Path: src/main/java/com/conductor/kafka/zk/ZkUtils.java
// @VisibleForTesting
// static class StringSerializer implements ZkSerializer {
// 
//     public StringSerializer() {
//     }
// 
//     @Override
//     public Object deserialize(byte[] data) throws ZkMarshallingError {
//         if (data == null)
//             return null;
//         return new String(data);
//     }
// 
//     @Override
//     public byte[] serialize(Object data) throws ZkMarshallingError {
//         return data.toString().getBytes();
//     }
// 
// }


// Path: src/test/java/com/conductor/kafka/zk/ZkUtilsTest.java
import com.conductor.kafka.Broker;
import com.conductor.kafka.Partition;
import com.conductor.kafka.zk.ZkUtils.StringSerializer;
import com.google.common.collect.Lists;
import org.I0Itec.zkclient.ZkClient;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import java.util.List;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;

        // with default namespace
        assertEquals("", zk.getZkRoot());

        // non-default namespace, tailing "/"
        zk = new ZkUtils(client, "/com/conductor/");
        assertEquals("/com/conductor", zk.getZkRoot());
    }

    @Test
    public void testGetBrokerId() throws Exception {
        // normal case
        final int brokerId = 1;
        when(client.readData("/brokers/ids/" + brokerId, true)).thenReturn("{ \"host\":\"127.0.0.1\", \"jmx_port\":9999, \"port\":9092, \"version\":1 }");
        final Broker broker = zk.getBroker(brokerId);
        assertEquals("127.0.0.1", broker.getHost());
        assertEquals(9092, broker.getPort());
        assertEquals(brokerId, broker.getId());

        // non-existant broker
        when(client.readData("/brokers/ids/" + 2, true)).thenReturn(null);
        final Broker nonExistantBroker = zk.getBroker(2);
        assertNull(nonExistantBroker);
    }

    @Test
    public void testGetBrokers() throws Exception {
        final List<String> brokerIds = Lists.newArrayList("1", "2");
        final Broker b1 = new Broker("localhost", 9092, 1);
        final Broker b2 = new Broker("localhost", 9092, 2);

        when(client.getChildren("/brokers/ids")).thenReturn(brokerIds);