package org.darkphoenixs.kafka.core;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import kafka.admin.TopicCommand;
import kafka.server.KafkaConfig;
import kafka.server.KafkaServer;
import kafka.utils.MockTime;
import kafka.utils.TestUtils;
import kafka.utils.TestZKUtils;
import kafka.utils.Time;
import kafka.utils.ZKStringSerializer$;
import kafka.zk.EmbeddedZookeeper;

import org.I0Itec.zkclient.ZkClient;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

public class ZookeeperBrokersTest {

	private int brokerId = 0;
	private String topic = "QUEUE.TEST";
	private String zkConnect;
	private EmbeddedZookeeper zkServer;
	private ZkClient zkClient;
	private KafkaServer kafkaServer;
	private int port;
	private Properties kafkaProps;

	@Before
	public void before() {
		zkConnect = TestZKUtils.zookeeperConnect();
		zkServer = new EmbeddedZookeeper(zkConnect);
		zkClient = new ZkClient(zkServer.connectString(), 30000, 30000,
				ZKStringSerializer$.MODULE$);

		// setup Broker
		port = TestUtils.choosePort();
		kafkaProps = TestUtils.createBrokerConfig(brokerId, port, true);

		KafkaConfig config = new KafkaConfig(kafkaProps);
		Time mock = new MockTime();
		kafkaServer = TestUtils.createServer(config, mock);

		// create topic
		TopicCommand.TopicCommandOptions options = new TopicCommand.TopicCommandOptions(
				new String[] { "--create", "--topic", topic,
						"--replication-factor", "1", "--partitions", "4" });

		TopicCommand.createTopic(zkClient, options);

		List<KafkaServer> servers = new ArrayList<KafkaServer>();
		servers.add(kafkaServer);
		TestUtils.waitUntilMetadataIsPropagated(
				scala.collection.JavaConversions.asScalaBuffer(servers), topic,
				0, 5000);
	}

	@After
	public void after() {
		kafkaServer.shutdown();
		zkClient.close();
		zkServer.shutdown();
	}

	@Test
	public void test() throws Exception {

		ZookeeperHosts zooHosts = new ZookeeperHosts(zkServer.connectString(),
				topic);

		final ZookeeperBrokers brokers = new ZookeeperBrokers(zooHosts);

		brokers.brokerPath();

		brokers.partitionPath();

		Assert.assertEquals(
				"localhost:9092",
				brokers.getBrokerHost("{\"host\":\"localhost\", \"jmx_port\":9999, \"port\":9092, \"version\":1 }"
						.getBytes("UTF-8")));

		try {
			brokers.getBrokerHost("1".getBytes());
		} catch (Exception e) {
			Assert.assertNotNull(e);
		}

		Assert.assertEquals(0, brokers.getLeaderFor(0));

		Assert.assertEquals(4, brokers.getNumPartitions());

		System.out.println(brokers.getBrokerInfo());

		brokers.close();
	}
}

@@ -2130,24 +2130,72 @@
 ionPath();%0A%0A
+%09%09System.out.println(brokers.getBrokerInfo());%0A%0A
 %09%09Assert.ass
@@ -2526,82 +2526,247 @@
 %0A%0A%09%09
-Assert.assertEquals(4, brokers.getNumPartitions());%0A%0A%09%09System.out.println(
+try %7B%0A%09%09%09brokers.getLeaderFor(-1);%0A%09%09%7D catch (Exception e) %7B%0A%09%09%09Assert.assertNotNull(e);%0A%09%09%7D%0A%0A%09%09Assert.assertEquals(4, brokers.getNumPartitions());%0A%0A%09%09try %7B%0A%09%09%09brokers.close();%0A%09%09%09brokers.getLeaderFor(0);%0A%09%09%09brokers.getNumPartitions();%0A%09%09%09
 brok
@@ -2788,30 +2788,68 @@
 fo()
-)
 ;%0A
-%0A%09%09brokers.close();
+%09%09%7D catch (Exception e) %7B%0A%09%09%09Assert.assertNotNull(e);%0A%09%09%7D%0A
 %0A%09%7D%0A

fix test