public interface KafkaConsumer {
	/**
	 * get consumer id which register to zookeeper.
	 * 
	 * @return
	 */
	public ConsumerId getConsumerId() ;

	/**
	 * handle all the partition reader exception.
	 */
	void handlePartitionReaderException(PartitionReader reader,KafkaPartitionReaderException e);
	
	/**
	 * do actrual rebalance.
	 */
	void rebalance();
	/**
	 * start ConsumerTask
	 */
	void start();
	/**
	 * stop consumerTask
	 */
	void stop();
	/**
	 * 
	 * @return
	 */
	boolean isRunning();
	/**
	 * commit offset to zookeeper.
	 */
	void commitOffset() throws Exception;
	/**
	 * mark commit offset.
	 * 
	 */
	void markCommitOffset();
	
	public void resetOffset(String topic, String group, int partition,long offset);
	public boolean takePartition(String topic,int partition) throws Exception;
	public void releasePartition(String topic,int partition);
	public String getTopic();
} public interface PartitionCoordinatorStrategy {
	/**
	 * Init PartitionCoordinatorStrategy implementor by providing config and AssignorContext.
	 * Usually implementor should keep the config and AssignorContext.
	 * 
	 * @param config
	 * @param ctx
	 */
	public void init(KafkaConsumerConfig config, String topic,String groupId,String consumerId) ;
	/**
	 * do coordinate.
	 * what coordinate will do are below:
	 * 
	 * 1. check whether there are over-taken partitions.
	 *     release them if exists.
	 * 2. check whether there are un-taken partitions
	 * 		take them according to the strategy.
	 * 
	 * @param ctx
	 */
	public void coordinate(AssignorContext ctx);
	/**
	 * after the coordinator takes a partition, do some extra process for the partition.
	 * e.g.: save this partition for extra use.
	 * 
	 * @param topic
	 * @param partition
	 */
	public void afterTakePartition(String topic, int partition);
	/**
	 * when the coordinator want to take a partition,
	 * do some extra process for this partition 
	 * once this return true, the coordinator will take the partition
	 * once this return false, the coordinator give up taking the partition.
	 * 
	 * @param topic
	 * @param partition
	 * @return
	 */
	public boolean beforeTakePartition(String topic,int partition);
} public interface PartitionReader extends Comparable<PartitionReader>{
	public PartitionReader startFrom(long offset);
	
	public void reinit();
	/**
	 * 	read events.
	 * 
	 * any errors occurred druing the read process are wrapped as KafkaPartitionReaderException which contains the error code
	 * the exception should be processed by consumer.
	 * 
	 * @return
	 * @throws KafkaPartitionReaderException
	 */
	public List<MessageAndMetadata<byte[],byte[]>> readEvents() throws KafkaPartitionReaderException ;
	public boolean readToTheEnd() ;
	public long getLargestOffset() ;
	public long fetchSmallestOffset();
	public long fetchResetOffset(String reset);
	public void commitOffset() throws Exception ;
	public void markCommitOffset();
	public long getBatchSize();
	public void initOffset() throws Exception; 
	public void resetOffset() throws Exception;
	public long getAndSetCommitOffset(long val);
	public boolean isTaken();
	public void release() ;
	public long calcWaitTime() ;
	public long getReadOffset();
	public void setReadOffset(long readOffset);
	public String getGroupId();
	public String getClientId();
	public String getTopic();
	public int getPartition();
	public void onIdle();
	public boolean isOnIdled() ;
	public void clearIdleStats() ;
	public void close() throws Exception ;
	public boolean isClosed();
	public void lost() ;
	public boolean isLost();
} public interface SimpleController {
	/**
	 * create a consumer for topic
	 * 
	 * @param topic
	 * @return
	 */
	public KafkaConsumer createConsumer(String topic);
	/**
	 * Get underlying zookeeper client.
	 * 
	 * @return
	 */
	public ZKConnector<?> getZkConnector() ;
	/**
	 * stop the controller.
	 * once it is stopped, it couldn't do anything unless started again.
	 */
	public void stop() ;
	/**
	 * start the controller. 
	 * once the controller started, it will get ready to do all things. for example, re-balance checking.
	 * 
	 * @throws InterruptedException
	 */
	public void start() throws InterruptedException;
	/**
	 * true after start() invoke successfully.
	 * 
	 * @return
	 */
	public boolean isStarted();
	/**
	 * register the consumer to zookeeper.
	 * 
	 * @param name
	 * @param consumer
	 */
	public void register(String name, KafkaConsumer consumer) ;
	/**
	 * un-register the consumer from zookeeper.
	 * @param name
	 */
	public void unregister(String name);
	/**
	 * 
	 * @return
	 */
	public boolean isRebalanceable() ;
	
} public interface ZKConnector<T> {
	/**
	 * Init the underlying zookeeper client, make sure it is connected.
	 * init the zookeeper state change listener and processor,
	 * especially for CONNECTED, DISCONNECTED, RECONNECTED, SESSION_EXPIRED states or states equivalent underlying framework.
	 * 
	 * @throws InterruptedException
	 */
	public void start() throws InterruptedException ;
	/**
	 * close zookeeper channel. 
	 * Cannot communicate with zookeeper any more unless <code>init()</code> invoked again and <code>isConnected()</code> return true;
	 * once close() invoked, isZkConnected() will return false;
	 */
	public void close() ;
	/**
	 * return current ZKConnector state, connected or disconnected.
	 * @return true is connected and false for disconnected.
	 */
	public boolean isZkConnected() ;
	/**
	 * return underlying zookeeper framework client.
	 * @return
	 */
	public T getZkClient();
	
	/**
	 * 
	 * make sure a persistent path exists in ZK. Create the path if not exist.
	 * Never throw NoNodeException or NodeExistsException.
	 * 
	 * @throws Exception 
	 */
	public void makeSurePersistentPathExists(String path) throws Exception ;

	/**
	 * create the parent path
	 * @throws Exception 
	 */
	public void createParentPath(String path) throws Exception ;
	/**
	 * Create an ephemeral node with the given path and data. Create parents if
	 * necessary.
	 * @throws Exception 
	 */
	public String createEphemeralPath(String path,
			String data) throws Exception ;
	/**
	 * Create an persistent node with the given path and data. Create parents if
	 * necessary.
	 * @throws Exception 
	 */
	public void createPersistentPath(String path,
			String data) throws Exception ;
	
	public String createSequentialPersistentPath(String path, String data) throws Exception ;
	/**
	 * Update the value of a persistent node with the given path and data.
	 * create parrent directory if necessary. 
	 * Never throw NodeExistException.
	 * @throws Exception 
	 */
	public void updateEphemeralPath(String path,
			String data) throws Exception ;
	/**
	 * Update the value of a persistent node with the given path and data.
	 * create parrent directory if necessary. Never throw NodeExistException.
	 * Return the updated path zkVersion
	 * @throws Exception 
	 */
	public void updatePersistentPath(String path,
			String data) throws Exception;
	/**
	 * delete path.
	 * Never throw NoNodeException.
	 * 
	 * @param path
	 * @return true is success and false is failed.
	 */
	public boolean deletePath(String path) throws Exception;
	/**
	 * delete path recursively. 
	 * Never throw NoNodeException.
	 * 
	 * @param path
	 */
	public void deletePathRecursive(String path) throws Exception;
	/**
	 * read data from path.
	 * Exception thrown based on underlying zookeeper client. 
	 * @param path
	 * @return
	 */
	public String readData(String path) throws Exception;
	/**
	 * read data from path, Pair.lhs is the String type data and Pair.rhs is the stat. 
	 * Never throw NoNodeException
	 * 
	 * @param path
	 * @return Pair.lhs may be null and the stat will be no use, otherwise return the data and stat pair.
	 */
	public Pair<String,Stat> readDataMaybeNullWithStat(String path) throws Exception;
	/**
	 * read data from path. 
	 * Never throw NoNodeException
	 * 
	 * @param path
	 * @return may be null if no data or no node exist.
	 */
	public String readDataMaybeNull(String path) throws Exception;
	/**
	 * get children for path. 
	 * Exception thrown based on underlying zookeeper client. 
	 * 
	 * @param path
	 * @return
	 */
	public List<String> getChildren(String path) throws Exception;
	/**
	 * get children for path, if parent deos not exist, return null, never throw NoNodeException.
	 *  
	 * @param path
	 * @return
	 */
	public List<String> getChildrenParentMayNotExist(String path) throws Exception;

	/**
	 * Check if the given path exists
	 * @throws Exception 
	 */
	public boolean pathExists(String path) throws Exception;
	
} public final class ConsumerId {
	private String consumerGroupId;
	private String hostName;
	private String consumerUuid;
	public ConsumerId(String consumerGroupId) {
		this(consumerGroupId, null, null);
	}

	public ConsumerId(String consumerGroupId, String uniqueId) {
		this(consumerGroupId, null, uniqueId);
	}

	public ConsumerId(String consumerGroupId, String hostName, String uniqueId) {
		checkArgument(consumerGroupId!=null,"consumerGroupId can't be null.");
		this.consumerGroupId = consumerGroupId;
		UUID uuid = UUID.randomUUID();
		if (hostName == null) {
			try {
				InetAddress s = InetAddress.getLocalHost();
				this.hostName = s.getHostName();
			} catch (UnknownHostException e) {
				throw new RuntimeException("Can resolve local host name", e);
			}
		}
		 consumerUuid = String.format("%s-%d-%s",
			        this.hostName, System.currentTimeMillis(),
			        Long.toHexString(uuid.getMostSignificantBits()).substring(0,8));
	}

	public String getConsumerGroupId() {
		return consumerGroupId;
	}

	public void setConsumerGroupId(String consumerGroupId) {
		this.consumerGroupId = consumerGroupId;
	}

	public String getHostName() {
		return hostName;
	}

	public void setHostName(String hostName) {
		this.hostName = hostName;
	}
	public String getConsumerUuid() {
		return consumerUuid;
	}
	public void setConsumerUuid(String consumerUuid) {
		this.consumerUuid = consumerUuid;
	}

	@Override
	public String toString() {
		return consumerGroupId+"_"+consumerUuid;
	}

	@Override
	public int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result
				+ ((consumerGroupId == null) ? 0 : consumerGroupId.hashCode());
		result = prime * result
				+ ((consumerUuid == null) ? 0 : consumerUuid.hashCode());
		result = prime * result
				+ ((hostName == null) ? 0 : hostName.hashCode());
		return result;
	}

	@Override
	public boolean equals(Object obj) {
		if (this == obj)
			return true;
		if (obj == null)
			return false;
		if (getClass() != obj.getClass())
			return false;
		ConsumerId other = (ConsumerId) obj;
		if (consumerGroupId == null) {
			if (other.consumerGroupId != null)
				return false;
		} else if (!consumerGroupId.equals(other.consumerGroupId))
			return false;
		if (consumerUuid == null) {
			if (other.consumerUuid != null)
				return false;
		} else if (!consumerUuid.equals(other.consumerUuid))
			return false;
		if (hostName == null) {
			if (other.hostName != null)
				return false;
		} else if (!hostName.equals(other.hostName))
			return false;
		return true;
	}

	
} public class KafkaConsumerConfig extends ConsumerConfig{
	private int rebalanceableWaitInMs = 15000;
	private int retryTimes=3;
	private int sleepMsBetweenRetries=500;
	public KafkaConsumerConfig(Properties originalProps) {
		super(originalProps);
		rebalanceableWaitInMs=PropertiesUtils.getInt(originalProps, "rebalance.first.wait.ms", 15000);
		retryTimes=PropertiesUtils.getInt(originalProps, "zookeeper.retry.times", 3);
		sleepMsBetweenRetries=PropertiesUtils.getInt(originalProps, "zookeeper.retry.backoff.ms", 500);
	}
	public int getRebalanceableWaitInMs() {
		return rebalanceableWaitInMs;
	}

	public void setRebalanceableWaitInMs(int rebalanceableWaitInMs) {
		this.rebalanceableWaitInMs = rebalanceableWaitInMs;
	}
	public int retryTimes(){
		return retryTimes;
	}
	public int sleepMsBetweenRetries(){
		return sleepMsBetweenRetries;
	}
} public class KafkaZKData {
	public static String ConsumersPath = "/consumers";
	public static String BrokerIdsPath = "/brokers/ids";
	public static String BrokerTopicsPath = "/brokers/topics";
	public static String TopicConfigPath = "/config/topics";

	public static String TopicConfigChangesPath = "/config/changes";
	public static String ControllerPath = "/controller";
	public static String ControllerEpochPath = "/controller_epoch";
	public static String ReassignPartitionsPath = "/admin/reassign_partitions";
	public static String DeleteTopicsPath = "/admin/delete_topics";
	public static String PreferredReplicaLeaderElectionPath = "/admin/preferred_replica_election";

	public static ObjectMapper mapper = new ObjectMapper();
	public static TypeReference<Map<String, Object>> typeMap = new TypeReference<Map<String, Object>>() {
	};
	//public static ZKStringSerializer serializer=new ZKStringSerializer();

	public static String topicPath(String topic) {
		return BrokerTopicsPath + "/" + topic;
	}

	public static String topicPartitionsPath(String topic) {
		return topicPath(topic) + "/partitions";
	}

	public static String topicConfigPath(String topic) {
		return TopicConfigPath + "/" + topic;
	}

	public static String deleteTopicPath(String topic) {
		return DeleteTopicsPath + "/" + topic;
	}

	public static String topicPartitionPath(String topic, int partitionId) {
		return topicPartitionsPath(topic) + "/" + partitionId;
	}

	public static String topicPartitionLeaderAndIsrPath(String topic,
			int partitionId) {
		return topicPartitionPath(topic, partitionId) + "/" + "state";
	}

	public static String partitionOwnerPath(String group, String topic,
			int partition) {
		ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(group, topic);
		return topicDirs.consumerOwnerDir() + "/" + partition;
	}
	public static Integer controller(ZKConnector<?> zkClient) {
		try{
			String data= zkClient.readDataMaybeNull(ControllerPath);
			checkState(data!=null,"Controller data is empty.");
			Map<String, Object> jsonObj = mapper.reader(typeMap).readValue(data);
			checkState(jsonObj!=null,"");
			Object id = jsonObj.get("brokerid");
			checkState(id!=null,"broker id does not exist.");
			return (Integer)id;
		}catch(Exception e){
			 throw new KafkaZKException(String.format("Failed to get controller data"),e);
		}
	}
	public static List<String> sortedBrokerList(ZKConnector<?> zkClient) {
		try{
			List<String> list=zkClient.getChildren(BrokerIdsPath);
			Collections.sort(list);
			return list;
		}catch(Throwable t){
			throw new KafkaZKException(t);
		}
	}

	public static List<Broker> allBrokersInCluster(final ZKConnector<?> zkClient) {
		try{
			List<String> brokerIds = Ordering.natural().sortedCopy(zkClient.getChildrenParentMayNotExist(BrokerIdsPath));
			return FluentIterable.from(brokerIds).transform(new Function<String,Broker>(){
				@Override
				public Broker apply(String input) {
					Integer i = Integer.parseInt(input);
					return getBrokerInfo(zkClient, i);
				}
			}).filter(new Predicate<Broker>() {
				@Override
				public boolean apply(Broker input) {
					return input != null;
				}
			}).toList();
		}catch(Exception e){
			return Lists.newArrayList();
		}
		
	}

	public static long getOffset(ZKConnector<?> zkClient, String path) {
		try {
			if (!zkClient.pathExists(path))
				return -1;
			String offsetStr = zkClient.readData(path);
			if (offsetStr == null)
				return -1;
			return Long.parseLong(offsetStr);
		} catch (Throwable t) {
			throw new KafkaZKException(t);
		} 
	}

	public static Set<Integer> allPartitionsOfTopic(ZKConnector<?> client,
			String topic) {
		try{
			Set<Integer> partitions = Sets.newHashSet();//
			String path = topicPartitionsPath(topic);
			List<String> pstrs = client.getChildrenParentMayNotExist(path);
			if (pstrs == null || pstrs.isEmpty()) {
				return partitions;
			}
			return ImmutableSet.copyOf(Iterables.transform(pstrs,
					new Function<String, Integer>() {
						@Override
						public Integer apply(String input) {
							return Integer.parseInt(input);
						}
				}));
		}catch(Throwable t){
			throw new KafkaZKException(t);
		}
	}

//	 def getConsumersPerTopic(zkClient: ZkClient, group: String, excludeInternalTopics: Boolean) : mutable.Map[String, List[ConsumerThreadId]] = {
//			    val dirs = new ZKGroupDirs(group)
//			    val consumers = getChildrenParentMayNotExist(zkClient, dirs.consumerRegistryDir)
//			    val consumersPerTopicMap = new mutable.HashMap[String, List[ConsumerThreadId]]
//			    for (consumer <- consumers) {
//			      val topicCount = TopicCount.constructTopicCount(group, consumer, zkClient, excludeInternalTopics)
//			      for ((topic, consumerThreadIdSet) <- topicCount.getConsumerThreadIdsPerTopic) {
//			        for (consumerThreadId <- consumerThreadIdSet)
//			          consumersPerTopicMap.get(topic) match {
//			            case Some(curConsumers) => consumersPerTopicMap.put(topic, consumerThreadId :: curConsumers)
//			            case _ => consumersPerTopicMap.put(topic, List(consumerThreadId))
//			          }
//			      }
//			    }
//			    for ( (topic, consumerList) <- consumersPerTopicMap )
//			      consumersPerTopicMap.put(topic, consumerList.sortWith((s,t) => s < t))
//			    consumersPerTopicMap
//			  }

	
	public static Map<String, Set<Integer>> partitionOwnerViewOfTopic(
			final ZKConnector<?> zkClient, String topic, String groupId) {
		Map<String, Set<Integer>> partitionOwnerView = Maps.newHashMap();
		try{
			String path = topicPartitionsPath(topic);
			List<String> pstrs = zkClient.getChildrenParentMayNotExist(path);
			if (pstrs == null || pstrs.isEmpty()) {
				return partitionOwnerView;
			}
			for (String pn : pstrs) {
				ZKGroupTopicDirs gt = new ZKGroupTopicDirs(groupId, topic);
				String ownerId = zkClient.readDataMaybeNull(gt.consumerOwnerDir()
						+ "/" + pn);
				if (ownerId != null) {
					Set<Integer> set = partitionOwnerView.get(ownerId);
					if (set == null) {
						set = Sets.newHashSet();
						partitionOwnerView.put(ownerId, set);
					}
					set.add(Integer.parseInt(pn));
				}
			}
		}catch(Throwable t){
			throw new KafkaZKException(t);
		}
		return partitionOwnerView;
	}

	public static Integer leaderForPartition(ZKConnector<?> zkClient, String topic,
			int partition) {
		try {
			String leaderAndIsrOpt = zkClient.readDataMaybeNull(topicPartitionLeaderAndIsrPath(topic, partition));
			checkNotNull(leaderAndIsrOpt);
			Map<String, Object> jsonObj = mapper.reader(typeMap).readValue(
					leaderAndIsrOpt);
			checkNotNull(jsonObj);
			Object leader = jsonObj.get("leader");
			checkNotNull(leader);
			return Integer.parseInt(leader.toString());
		} catch (Exception e) {
			return null;
		}
	}

	@SuppressWarnings("unchecked")
	public static LeaderIsrAndControllerEpoch leaderIsrAndEpochForPartition(
			ZKConnector<?> zkClient, String topic, int partition) {
		
		try{
			String leaderAndIsrPath = topicPartitionLeaderAndIsrPath(topic, partition);
			Pair<String,Stat> leaderAndIsrInfo =zkClient.readDataMaybeNullWithStat(leaderAndIsrPath);
			Map<String, Object> jsonObj = mapper.reader(typeMap).readValue(leaderAndIsrInfo.lhs);
			Integer leader=(Integer)jsonObj.get("leader");
			Integer epoch=(Integer)jsonObj.get("leader_epoch");
			List<Object> isr=(List<Object>)jsonObj.get("isr");
			Integer controllerEpoch=(Integer)jsonObj.get("controller_epoch");
			return new LeaderIsrAndControllerEpoch(new LeaderAndIsr(leader,epoch,(scala.collection.immutable.List<Object>) isr,leaderAndIsrInfo.rhs.getVersion()),controllerEpoch);
		}catch(Exception e){
			
		}
		return  null;
	}

	public static LeaderAndIsr leaderAndIsrForPartition(ZKConnector<?> zkClient,
			String topic, int partition) {
		LeaderIsrAndControllerEpoch entity = leaderIsrAndEpochForPartition(
				zkClient, topic, partition);
		if (entity != null)
			return entity.leaderAndIsr();
		return null;
	}

	public static void registerConsumer(ZKConnector<?> zkClient, String topic,
			String groupId, String consumerId) {
		ZKGroupTopicDirs dirs = new ZKGroupTopicDirs(groupId, topic);
		TopicCount topicCount = TopicCount.constructTopicCount(consumerId,
				ImmutableMap.of(topic, 1));
		String timestamp = String.valueOf(System.currentTimeMillis());
		ImmutableMap<String, Object> data = ImmutableMap.of("version", 1, "subscription",
				topicCount.getTopicCountMap(), "pattern", topicCount.pattern(),
				"timestamp", timestamp);
		String consumerRegistrationInfo = JSON.toJSONString(data);

		createEphemeralPathExpectConflictHandleZKBug(zkClient,
				dirs.consumerRegistryDir() + "/" + consumerId,
				consumerRegistrationInfo, consumerId,
				new Function<Pair<String, Object>, Boolean>() {
					@Override
					public Boolean apply(Pair<String, Object> input) {
						return true;
					}
				}, 5000);
	}
	public static void unregisterConsumer(ZKConnector<?> zkClient,String topic,String groupId,String consumerId){
		try{
		ZKGroupTopicDirs dirs = new ZKGroupTopicDirs(groupId, topic);
		zkClient.deletePath(dirs.consumerRegistryDir()+"/"+consumerId);
		Map<String,Set<Integer>> view=KafkaZKData.partitionOwnerViewOfTopic(zkClient, topic, groupId);
		Set<Integer> partitions=view.get(consumerId);
		if(partitions!=null && partitions.size()>0){
			for(Integer p: partitions){
				releasePartition(zkClient,topic,groupId,p,consumerId);
			}
		}
		}catch(Throwable t){
			throw new KafkaZKException(t);
		}
	}

	public static boolean takePartition(ZKConnector<?> zkClient, String topic,
			String groupId, int partition, String consumerId) {
		try {
			ZKGroupTopicDirs dirs = new ZKGroupTopicDirs(groupId, topic);
			zkClient.makeSurePersistentPathExists(dirs.consumerOwnerDir());
			// TODO: check the consumer register infor
			//throw new RuntimeException("Debug.");
			KafkaZKData.createEphemeralPathExpectConflictHandleZKBug(zkClient,
					dirs.consumerOwnerDir() + "/" + partition, consumerId,
					consumerId, new Function<Pair<String,Object>,Boolean>(){
						@Override
						public Boolean apply(Pair<String, Object> input) {
							return input.rhs.equals(input.lhs);
						}
				
			}, 5000);
			return true;
		} catch (Throwable t) {
			return false;
		}
	}

	public static void releasePartition(ZKConnector<?> zkClient, String topic,
			String groupId, int partition, String consumerId) {
		ZKGroupTopicDirs dirs = new ZKGroupTopicDirs(groupId, topic);
		String ownerPath = dirs.consumerOwnerDir() + "/" + partition;
		try{
			if (zkClient.pathExists(ownerPath)
					&& consumerId.equals(zkClient.readData(ownerPath))) {
				zkClient.deletePath(ownerPath);
			}
		}catch(Exception e){
			e.printStackTrace();
			//exception when communicate zookeeper, return normally.
		}
	}

	public static void setupCommonPaths(ZKConnector<?> zkClient) throws Exception {
		for (String path : ImmutableList.of(ConsumersPath, BrokerIdsPath,
				BrokerTopicsPath, TopicConfigChangesPath, TopicConfigPath,
				DeleteTopicsPath)) {
			zkClient.makeSurePersistentPathExists(path);
		}
	}

	/**
	 * This API should read the epoch in the ISR path. It is sufficient to read
	 * the epoch in the ISR path, since if the leader fails after updating epoch
	 * in the leader path and before updating epoch in the ISR path, effectively
	 * some other broker will retry becoming leader with the same new epoch
	 * value.
	 * @throws Exception 
	 */
	public static Integer epochForPartition(ZKConnector<?> zkClient, String topic,
			int partition) {
		LeaderAndIsr isr = leaderAndIsrForPartition(zkClient, topic, partition);
		if (isr != null)
			return isr.leaderEpoch();
		else
			throw new KafkaZKException(String.format(
					"No epoch, ISR path for partition [%s,%d] is empty", topic,
					partition));
	}

	@SuppressWarnings("unchecked")
	public static List<Integer> isrForPartition(ZKConnector<?> zkClient,
			String topic, int partition) {
		LeaderAndIsr isr = leaderAndIsrForPartition(zkClient, topic, partition);
		if (isr == null)
			throw new KafkaZKException(
					String.format(
							"No epoch, leaderAndISR data for partition [%s,%d] is invalid",
							topic, partition));
		return (List<Integer>) isr.isr();
	}

	public static List<Integer> assignedReplicasForPartition(ZKConnector<?> zkClient,
			String topic, int partition) {
		return assignedReplicasView(zkClient, topic).get(
				String.valueOf(partition));
	}

	@SuppressWarnings("unchecked")
	public static Map<String, List<Integer>> assignedReplicasView(
			ZKConnector<?> zkClient, String topic) {
		try {
			String jsonPartitionMapOpt = zkClient.readDataMaybeNull(	topicPath(topic));
			checkNotNull(jsonPartitionMapOpt);
			Map<String, Object> jsonObj = mapper.reader(typeMap).readValue(
					jsonPartitionMapOpt);
			checkNotNull(jsonObj);
			Map<String, List<Integer>> partitions = (Map<String, List<Integer>>) jsonObj
					.get("partitions");
			checkNotNull(partitions);
			return partitions;
		} catch (Exception e) {
			return Maps.newHashMap();
		}
	}

	public static String leaderAndIsrZkData(LeaderAndIsr leaderAndIsr,
			int controllerEpoch) {
		return Json.encode(ImmutableMap.of("version", 1, "leader",
				leaderAndIsr.leader(), "leader_epoch",
				leaderAndIsr.leaderEpoch(), "controller_epoch",
				controllerEpoch, "isr", leaderAndIsr.isr()));
	}

	/**
	 * Get JSON partition to replica map from zookeeper.
	 */
	public static String replicaAssignmentZkData(Map<String, List<Integer>> map) {
		return Json.encode(ImmutableMap.of("version", 1, "partitions", map));
	}

	/**
	 * Create an ephemeral node with the given path and data. Throw
	 * NodeExistException if node already exists.
	 * @throws NodeExistsException 
	 * @throws Throwable 
	 */
	public static void createEphemeralPathExpectConflict(ZKConnector<?> client,
			String path, String data) throws NodeExistsException{
		try {
			client.createEphemeralPath(path, data);
		} catch ( KeeperException.NodeExistsException e) {
			// this can happen when there is connection loss; make sure the data
			// is what we intend to write
			String storedData = null;
			try {
				storedData = client.readData(path);
			}catch (Throwable t) {
				throw new KafkaZKException(t);
			}
			if (storedData == null || storedData != data) {
				// info("conflict in " + path + " data: " + data +
				// " stored data: " + storedData)
				throw  e ;
			} else {
				// otherwise, the creation succeeded, return normally
				// info(path + " exists with value " + data +
				// " during connection loss; this is ok")
			}
		} catch (Throwable t) {
			throw new KafkaZKException(t);
		}
	}

	/**
	 * Create an ephemeral node with the given path and data. Throw
	 * NodeExistsException if node already exists. Handles the following ZK
	 * session timeout bug:
	 *
	 * https://issues.apache.org/jira/browse/ZOOKEEPER-1740
	 *
	 * Upon receiving a NodeExistsException, read the data from the conflicted
	 * path and trigger the checker function comparing the read data and the
	 * expected data, If the checker function returns true then the above bug
	 * might be encountered, back off and retry; otherwise re-throw the
	 * exception
	 * @throws NodeExistsException 
	 * @throws Exception 
	 */
	public static void createEphemeralPathExpectConflictHandleZKBug(
			ZKConnector<?> zkClient, String path, String data,
			Object expectedCallerData,
			Function<Pair<String, Object>, Boolean> checker, int backoffTime) {
				
		while (true) {
			try {
				createEphemeralPathExpectConflict(zkClient, path, data);
				return;
			} catch (KeeperException.NodeExistsException e) {
				// An ephemeral node may still exist even after its
				// corresponding session has expired
				// due to a Zookeeper bug, in this case we need to retry writing
				// until the previous node is deleted
				// and hence the write succeeds without ZkNodeExistsException
				try{
				String writtenData = zkClient.readDataMaybeNull(path);
				if (writtenData != null) {
					if (checker.apply(Pair.of(writtenData, expectedCallerData))) {
						// info("I wrote this conflicted ephemeral node [%s] at %s a while back in a different session, ".format(data,
						// path)
						// +
						// "hence I will backoff for this node to be deleted by Zookeeper and retry");
						try {
							zkClient.deletePath(path);
							Thread.sleep(backoffTime);
						} catch (Throwable e1) {
							e1.printStackTrace();
						}
					} else {
						throw new KafkaZKException(e);
					}
				}
				}catch(Throwable t){
					throw new KafkaZKException(t);
				}

			} catch (Throwable t) {
				throw new KafkaZKException(t);
			}
		}
	}

//
//	/**
//	 * Conditional update the persistent path data, return (true, newVersion) if
//	 * it succeeds, otherwise (the path doesn't exist, the current version is
//	 * not the expected version, etc.) return (false, -1)
//	 *
//	 * When there is a ConnectionLossException during the conditional update,
//	 * zkClient will retry the update and may fail since the previous update may
//	 * have succeeded (but the stored zkVersion no longer matches the expected
//	 * one). In this case, we will run the optionalChecker to further check if
//	 * the previous write did indeed succeeded.
//	 */
//	public static Pair<Boolean, Integer> conditionalUpdatePersistentPath(
//			ZKConnector<?> zkClient,
//			String path,
//			String data,
//			int expectVersion,
//			Function<Triple<ZKConnector<?>, String, String>, Pair<Boolean, Integer>> checker) {
//		try {
//			Stat stat=zkClient.setData().withVersion(expectVersion).forPath(path, serializer.serialize(data));
//			//Stat stat = zkClient.writeDataReturnStat(path, data, expectVersion);
//			return Pair.of(true, stat.getVersion());
//		} catch (KeeperException.BadVersionException e1) {
//			if (checker != null)
//				checker.apply(Triple.of(zkClient, path, data));
//			else {
//				// log warn:
//				// debug("Checker method is not passed skipping zkData match")
//				// warn(String.format("Conditional update of path %s with data %s and expected version %d failed due to %s",path,
//				// data, expectVersion, e1.getMessage))
//				return Pair.of(false, -1);
//			}
//		} catch (Exception e2) {
//			// warn(String.format("Conditional update of path %s with data %s and expected version %d failed due to %s",path,
//			// data,expectVersion, e2.getMessage))
//			return Pair.of(false, -1);
//		}
//		return null;
//
//	}

//	/**
//	 * Conditional update the persistent path data, return (true, newVersion) if
//	 * it succeeds, otherwise (the current version is not the expected version,
//	 * etc.) return (false, -1). If path doesn't exist, throws ZkNoNodeException
//	 * @throws NoNodeException 
//	 */
//	public static Pair<Boolean, Integer> conditionalUpdatePersistentPathIfExists(
//			ZKConnector<?> zkClient, String path, String data, int expectVersion) throws NoNodeException {
//		try {
//			Stat stat=zkClient.setData().withVersion(expectVersion).forPath(path, serializer.serialize(data));
//			//Stat stat = zkClient.writeDataReturnStat(path, data, expectVersion);
//			return Pair.of(true, stat.getVersion());
//		} catch (KeeperException.NoNodeException nne) {
//			throw nne;
//		} catch (Exception e) {
//			// error(String.format("Conditional update of path %s with data %s and expected version %d failed due to %s",path,
//			// data,expectVersion, e.getMessage))
//			return Pair.of(false, -1);
//		}
//	}

	

	public static Cluster getCluster(ZKConnector<?> zkClient) {
		try{
			Cluster cluster = new Cluster();
			List<String> nodes = zkClient.getChildrenParentMayNotExist(BrokerIdsPath);
			for (String node : nodes) {
				String brokerZKString = zkClient.readData( BrokerIdsPath + "/"	+ node);
				cluster.add(Broker.createBroker(Integer.parseInt(node),
						brokerZKString));
			}
			return cluster;
		}catch(Throwable t){
			throw new KafkaZKException(t);
		}
	}

	public static Map<String, Map<Integer, List<Integer>>> getPartitionAssignmentForTopics(
			final ZKConnector<?> zkClient, List<String> topics) {
		return Maps.toMap(topics,
				new Function<String, Map<Integer, List<Integer>>>() {
					@SuppressWarnings("unchecked")
					@Override
					public Map<Integer, List<Integer>> apply(String topic) {
						try {
							String jsonPartitionMapOpt = zkClient.readDataMaybeNull(topicPath(topic));
							checkNotNull(jsonPartitionMapOpt);
							Map<String, Object> jsonObj = mapper
									.reader(typeMap).readValue(
											jsonPartitionMapOpt);
							checkNotNull(jsonObj);
							Map<String, List<Integer>> partitions = (Map<String, List<Integer>>) jsonObj
									.get("partitions");
							checkNotNull(partitions);
							Map<Integer, List<Integer>> tmp = Maps.newHashMap();
							for (Entry<String, List<Integer>> entity : partitions
									.entrySet()) {
								tmp.put(Integer.parseInt(entity.getKey()),
										entity.getValue());
							}
							return tmp;
						} catch (Exception e) {
							return Maps.newHashMap();
						}
					}
				});
	}

	public static Map<String, List<Integer>> getPartitionsForTopics(
			final ZKConnector<?> zkClient, final List<String> topics) {
		return Maps.toMap(topics, new Function<String, List<Integer>>() {
			@Override
			public List<Integer> apply(String topic) {
				return Ordering.natural().sortedCopy(
						getPartitionAssignmentForTopics(zkClient, topics).get(
								topic).keySet());
			}
		});
	}

	@SuppressWarnings("rawtypes")
	public static String getPartitionReassignmentZkData(
			Map<TopicAndPartition, List<Integer>> partitionsToBeReassigned) {
		return Json
				.encode(ImmutableMap.of(
						"version",
						1,
						"partitions",
						Lists.transform(
								Lists.newArrayList(partitionsToBeReassigned
										.entrySet()),
								new Function<Entry<TopicAndPartition, List<Integer>>, Map>() {
									@Override
									public Map apply(
											Entry<TopicAndPartition, List<Integer>> input) {
										return ImmutableMap.of("topic", input
												.getKey().topic(), "partition",
												input.getKey().partition(),
												"replicas", input.getValue());
									}
								})));
	}

	public static void updatePartitionReassignmentData(ZKConnector<?> zkClient,
			Map<TopicAndPartition, List<Integer>> partitionsToBeReassigned) {
		try{
		if (partitionsToBeReassigned.size() == 0) {// need to delete the
													// /admin/reassign_partitions
													// path
			zkClient.deletePath(ReassignPartitionsPath);
		} else {
			String jsonData = getPartitionReassignmentZkData(partitionsToBeReassigned);
			zkClient.updatePersistentPath(ReassignPartitionsPath, jsonData);
		}
		}catch(Throwable t){
			throw new KafkaZKException(t);
		}
	}

	@SuppressWarnings("unchecked")
	public static Set<TopicAndPartition> getPartitionsUndergoingPreferredReplicaElection(
			ZKConnector<?> zkClient) {
		try{
		// read the partitions and their new replica list
			String jsonPartitionListOpt = zkClient.readDataMaybeNull(PreferredReplicaLeaderElectionPath);
			Map<String, Object> jsonObj = mapper.reader(typeMap).readValue(	jsonPartitionListOpt);
			checkState(jsonObj!=null,String.format("Preferred replica election data.[%s]", jsonPartitionListOpt));
			List<Map<String,Object>> partitions=(List<Map<String,Object>>)jsonObj.get("partitions");
			checkState(partitions!=null,"Preferred replica election data is empty");
			List<TopicAndPartition> plist=Lists.transform(partitions, new Function<Map<String,Object>,TopicAndPartition>(){
				@Override
				public TopicAndPartition apply(Map<String, Object> input) {
					return new TopicAndPartition((String)input.get("topic"),(Integer)input.get("partition"));
				}
			});
			Set<TopicAndPartition> sets=ImmutableSet.copyOf(plist);
			checkState(sets.size()==plist.size(),"Preferred replica election data contains duplicate partitions");
			return sets;
		}catch(Throwable t){
			throw new KafkaZKException(t);
		}
	}

	public static void deletePartition(ZKConnector<?> zkClient, int brokerId,
			String topic) {
		try{
		String brokerIdPath = BrokerIdsPath + "/" + brokerId;
		zkClient.deletePath(brokerIdPath);
		String brokerPartTopicPath = BrokerTopicsPath + "/" + topic + "/"
				+ brokerId;
		zkClient.deletePath(brokerPartTopicPath);//.delete(brokerPartTopicPath);
		}catch(Throwable t){
			throw new KafkaZKException(t);
		}
	}

	public static List<String> getConsumersInGroup(ZKConnector<?> zkClient,
			String group) {
		try{
		ZKGroupDirs dirs = new ZKGroupDirs(group);
		return zkClient.getChildren(dirs.consumerRegistryDir());
		}catch(Throwable t){
			throw new KafkaZKException(t);
		}
	}

	public static List<String> consumersOfTopic(ZKConnector<?> zkClient,
			String group, String topic) {
		try{
		ZKGroupDirs dirs = new ZKGroupDirs(group);
		List<String> consumers = zkClient.getChildrenParentMayNotExist(dirs.consumerRegistryDir());
		Map<String, List<String>> consumersPerTopicMap = Maps.newHashMap();
		for (String consumer : consumers) {
			TopicCount topicCount = TopicCount.constructTopicCount(zkClient,
					group, consumer);
			for (Entry<String, Set<TopicCount.ConsumerThreadId>> entity : topicCount
					.getConsumerThreadIdsPerTopic().entrySet()) {
				for (TopicCount.ConsumerThreadId consumerId : entity.getValue()) {
					List<String> c = consumersPerTopicMap.get(entity.getKey());
					if (c == null) {
						c = Lists.newArrayList();
						consumersPerTopicMap.put(entity.getKey(), c);
					}
					c.add(consumerId.getConsumer());
				}
			}
		}
		return consumersPerTopicMap.get(topic);
		}catch(Exception t){
			throw new KafkaZKException(t);
		}
	}

	/**
	 * This API takes in a broker id, queries zookeeper for the broker metadata
	 * and returns the metadata for that broker or throws an exception if the
	 * broker dies before the query to zookeeper finishes
	 * 
	 * @param brokerId
	 *            The broker id
	 * @param zkClient
	 *            The zookeeper client connection
	 * @return An optional Broker object encapsulating the broker metadata
	 * @throws Exception 
	 */
	public static Broker getBrokerInfo(ZKConnector<?> zkClient, int brokerId) {
		try{
		String brokerInfo = zkClient.readDataMaybeNull(BrokerIdsPath + "/"	+ brokerId);
		if (brokerInfo != null)
			return Broker.createBroker(brokerId, brokerInfo);
		}catch(Throwable t){
		}
		return null;
	}

	public static List<String> getAllTopics(ZKConnector<?> zkClient) {
		try{
			return zkClient.getChildrenParentMayNotExist(BrokerTopicsPath);
		}catch(Throwable t){
			throw new KafkaZKException(t);
		}
	}

	public static Set<TopicAndPartition> getAllPartitions(
			final ZKConnector<?> zkClient) {
		try{
		List<String> topics = zkClient.getChildrenParentMayNotExist(BrokerTopicsPath);
		if(topics==null) return Sets.newHashSet();
		return Sets.newHashSet(Iterables.concat(Lists.transform(topics,
				new Function<String, Iterable<TopicAndPartition>>() {
					@Override
					public Iterable<TopicAndPartition> apply(
							final String topic) {
						try{
						List<String> partitions = zkClient.getChildren(topicPartitionsPath(topic));
						return Lists.transform(partitions,
								new Function<String, TopicAndPartition>() {
									@Override
									public TopicAndPartition apply(
											String input) {
										return new TopicAndPartition(topic,
												Integer.parseInt(input));
									}
								});
						}catch(Throwable t){
							
						}
						return Lists.newArrayList();
					}
				})));
		}catch(Throwable t){
			throw new KafkaZKException(t);
		}
	}

	public static class Triple<T1, T2, T3> {
		public static <T1, T2, T3> Triple<T1, T2, T3> of(T1 lhs, T2 mhs, T3 rhs) {
			return new Triple<>(lhs, mhs, rhs);
		}

		public final T1 lhs;
		public final T2 mhs;
		public final T3 rhs;

		public Triple(T1 lhs, T2 mhs, T3 rhs) {
			this.lhs = lhs;
			this.rhs = rhs;
			this.mhs = mhs;
		}

		@Override
		public int hashCode() {
			final int prime = 31;
			int result = 1;
			result = prime * result + ((lhs == null) ? 0 : lhs.hashCode());
			result = prime * result + ((mhs == null) ? 0 : mhs.hashCode());
			result = prime * result + ((rhs == null) ? 0 : rhs.hashCode());
			return result;
		}

		@Override
		public boolean equals(Object obj) {
			if (this == obj)
				return true;
			if (obj == null)
				return false;
			if (getClass() != obj.getClass())
				return false;
			Triple<?,?,?> other = (Triple<?,?,?>) obj;
			if (lhs == null) {
				if (other.lhs != null)
					return false;
			} else if (!lhs.equals(other.lhs))
				return false;
			if (mhs == null) {
				if (other.mhs != null)
					return false;
			} else if (!mhs.equals(other.mhs))
				return false;
			if (rhs == null) {
				if (other.rhs != null)
					return false;
			} else if (!rhs.equals(other.rhs))
				return false;
			return true;
		}

		@Override
		public String toString() {
			return "Triple{" + "lhs=" + lhs + ",mhs=" + mhs + ", rhs=" + rhs
					+ '}';
		}

		public static <T1, T2, T3> Function<Triple<T1, T2, T3>, T1> lhsFn() {
			return new Function<Triple<T1, T2, T3>, T1>() {
				@Override
				public T1 apply(Triple<T1, T2, T3> input) {
					return input.lhs;
				}
			};
		}

		public static <T1, T2, T3> Function<Triple<T1, T2, T3>, T2> mhsFn() {
			return new Function<Triple<T1, T2, T3>, T2>() {
				@Override
				public T2 apply(Triple<T1, T2, T3> input) {
					return input.mhs;
				}
			};
		}

		public static <T1, T2, T3> Function<Triple<T1, T2, T3>, T3> rhsFn() {
			return new Function<Triple<T1, T2, T3>, T3>() {
				@Override
				public T3 apply(Triple<T1, T2, T3> input) {
					return input.rhs;
				}
			};
		}
	}

	public static class Pair<T1, T2> {

		public static <T1, T2> Pair<T1, T2> of(T1 lhs, T2 rhs) {
			return new Pair<>(lhs, rhs);
		}

		public final T1 lhs;
		public final T2 rhs;

		public Pair(T1 lhs, T2 rhs) {
			this.lhs = lhs;
			this.rhs = rhs;
		}

		@Override
		public boolean equals(Object o) {
			if (this == o) {
				return true;
			}
			if (o == null || getClass() != o.getClass()) {
				return false;
			}

			Pair<?,?> pair = (Pair<?,?>) o;

			if (lhs != null ? !lhs.equals(pair.lhs) : pair.lhs != null) {
				return false;
			}
			if (rhs != null ? !rhs.equals(pair.rhs) : pair.rhs != null) {
				return false;
			}

			return true;
		}

		@Override
		public int hashCode() {
			int result = lhs != null ? lhs.hashCode() : 0;
			result = 31 * result + (rhs != null ? rhs.hashCode() : 0);
			return result;
		}

		@Override
		public String toString() {
			return "Pair{" + "lhs=" + lhs + ", rhs=" + rhs + '}';
		}

		public static <T1, T2> Function<Pair<T1, T2>, T1> lhsFn() {
			return new Function<Pair<T1, T2>, T1>() {
				@Override
				public T1 apply(Pair<T1, T2> input) {
					return input.lhs;
				}
			};
		}

		public static <T1, T2> Function<Pair<T1, T2>, T2> rhsFn() {
			return new Function<Pair<T1, T2>, T2>() {
				@Override
				public T2 apply(Pair<T1, T2> input) {
					return input.rhs;
				}
			};
		}

		public static <T1> Comparator<Pair<T1, ?>> lhsComparator(
				final Comparator<T1> comparator) {
			return new Comparator<Pair<T1, ?>>() {
				@Override
				public int compare(Pair<T1, ?> o1, Pair<T1, ?> o2) {
					return comparator.compare(o1.lhs, o2.lhs);
				}
			};
		}

	}

	public static class ZKGroupDirs {
		private String consumerDir = ConsumersPath;
		private String group;

		public ZKGroupDirs(String group) {
			this.group = group;
		}

		public String consumerGroupDir() {
			return consumerDir + "/" + group;
		}

		public String consumerRegistryDir() {
			return consumerGroupDir() + "/ids";
		}
	}

	public static class ZKGroupTopicDirs extends ZKGroupDirs {
		private String topic;

		public ZKGroupTopicDirs(String group, String topic) {
			super(group);
			this.topic = topic;
		}

		public String consumerOffsetDir() {
			return consumerGroupDir() + "/offsets/" + topic;
		}

		public String consumerOwnerDir() {
			return consumerGroupDir() + "/owners/" + topic;
		}

		public String partitionOffset(Integer partition) {
			return consumerOffsetDir() + "/" + partition;
		}
	}
//	public static class ZKStringSerializer {
//		public byte[] serialize(Object data) {
//			try {
//				return data.toString().getBytes("UTF-8");
//			} catch (UnsupportedEncodingException e) {
//				throw new KafkaZKException(e);
//			}
//		}
//		public String deserialize(byte[] bytes) {
//			if (bytes == null)
//				return null;
//			else
//				try {
//					return new String(bytes, "UTF-8");
//				} catch (UnsupportedEncodingException e) {
//					throw new KafkaZKException(e);
//				}
//		}
//	}
} public class KafkaPartitionReaderException extends RuntimeException {
	private static final long serialVersionUID = -5969989670100717318L;
	private short code;
	public KafkaPartitionReaderException(short code){
		super("KafkaPartition Reader Exception.code="+code);
		this.code=code;
	}
	public KafkaPartitionReaderException(short code,String msg){
		super(msg);
		this.code=code;
	}
	public short getCode(){
		return code;
	}
} public class KafkaZKException extends RuntimeException {
	private static final long serialVersionUID = -7507119817636549797L;
	public KafkaZKException(String msg){
		super(msg);
	}
	public KafkaZKException(Throwable t){
		super(t);
	}
	public KafkaZKException(String msg,Throwable t){
		super(msg,t);
	}
} public class LOG {
	private static final Logger log = new Logger(LOG.class);
	public static void log(String msg){
		log.info("EX[%d][%s]:%s", Thread.currentThread().getId(),Thread.currentThread().getName(),msg);
		//String s="EX["+Thread.currentThread().getId()+"]["+Thread.currentThread().getName()+"]:"+msg;
		//System.out.println(s);
		//log.info(s);
		
	}
	public static void log(String msg, Throwable t){
		//log.info("EX[%d][%s]:%s", Thread.currentThread().getId(),Thread.currentThread().getName(),msg);
		log.error(t, "EX[%d][%s]:%s", Thread.currentThread().getId(),Thread.currentThread().getName(),msg);
		//String s="EX["+Thread.currentThread().getId()+"]["+Thread.currentThread().getName()+"]:"+msg;
		//System.out.println(s);
		//t.printStackTrace();
		//log.info(s,t);
	}
} public class NameableThreadFactory implements ThreadFactory {

	final ThreadGroup group;
	final AtomicInteger threadNumber = new AtomicInteger(1);
	final String namePrefix;

	public NameableThreadFactory(final String name) {
		SecurityManager s = System.getSecurityManager();
		group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
		namePrefix = name + "-";
	}

	public Thread newThread(Runnable r) {
		Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
		if (t.isDaemon()) {
			t.setDaemon(false);
		}
		if (t.getPriority() != Thread.NORM_PRIORITY) {
			t.setPriority(Thread.NORM_PRIORITY);
		}
		return t;
	}

}
import static com.google.common.base.Preconditions.checkArgument;
import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.ClosedChannelException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import kafka.common.ErrorMapping;
import kafka.common.TopicAndPartition;
import kafka.message.MessageAndMetadata;
import org.apache.zookeeper.KeeperException.ConnectionLossException;
import com.ebay.pulsar.druid.firehose.kafka.support.api.KafkaConsumer;
import com.ebay.pulsar.druid.firehose.kafka.support.api.PartitionCoordinatorStrategy;
import com.ebay.pulsar.druid.firehose.kafka.support.api.PartitionReader;
import com.ebay.pulsar.druid.firehose.kafka.support.api.SimpleController;
import com.ebay.pulsar.druid.firehose.kafka.support.api.ZKConnector;
import com.ebay.pulsar.druid.firehose.kafka.support.data.ConsumerId;
import com.ebay.pulsar.druid.firehose.kafka.support.data.KafkaConsumerConfig;
import com.ebay.pulsar.druid.firehose.kafka.support.data.KafkaZKData;
import com.ebay.pulsar.druid.firehose.kafka.support.exceptions.KafkaPartitionReaderException;
import com.ebay.pulsar.druid.firehose.kafka.support.exceptions.KafkaZKException;
import com.ebay.pulsar.druid.firehose.kafka.support.util.LOG;
import com.ebay.pulsar.druid.firehose.kafka.support.util.NameableThreadFactory;
			lock.lock();
			try{
				while(queue.size()==0){
					//LOG.log("Await..."+count.getAndIncrement());
					notEmpty.await();
				}
			}catch(Exception e){
				e.printStackTrace();
			}finally{
				lock.unlock();
			}
		}
		@Override
		public boolean hasNext() {
			//await();
			lock.lock();
			try{
				while(queue.size()==0){
					notEmpty.await();
				}
				return queue.size()>0;
			}catch(Exception e){
				e.printStackTrace();
			}finally{
				lock.unlock();
			}
			//return queue.size()>0;
			return false;
		}
		@Override
		public MessageAndMetadata<byte[],byte[]> next() {
			readCount.incrementAndGet();
			lock.lock();
			try {
				while(queue.size()==0){
					notEmpty.await();
				}
				MessageAndMetadata<byte[], byte[]> t= queue.poll();//queue.take();
				notFull.signalAll();
				TopicAndPartition key = new TopicAndPartition(t.topic(), t.partition());
				PartitionReader reader=partitionMap.get(key);
				if(reader!=null)
				{
					commitCount.incrementAndGet();
					reader.getAndSetCommitOffset(t.offset());
					return t;
				}
			} catch (Exception e) {
				LOG.log("Read Message next() exception:",e);
			}finally{
				lock.unlock();
			}
			return null;
		}
		@Override
		public void remove() {
			throw new RuntimeException("Not support remove operation.");
		}
	}
	
	
	public class ConsumerTask implements Runnable {
		private String taskId;
		private LinkedList<PartitionReader> hold = new LinkedList<PartitionReader>();
		private List<MessageAndMetadata<byte[],byte[]>> cache = new ArrayList<MessageAndMetadata<byte[],byte[]>>();
		private PartitionReader borrowReadableReaderAndBlokingIfNotExist() throws InterruptedException{
			PartitionReader preader =queue.poll();
			do{
				if (preader == null) {
					if (hold.isEmpty()) {
						preader = queue.take();
					} else {
						PartitionReader earliest = hold.getFirst();
						long wait = earliest.calcWaitTime();
						if (wait > 0) {
							preader = queue.poll(wait,
									TimeUnit.MILLISECONDS);
						}
						if (preader == null)
							preader = hold.poll();
					}
				}else{
					if (preader.calcWaitTime() > 0) {
						hold(preader);
						preader = null;
					}
				}
			}while(preader==null);
			return preader;
		}

		@Override
		public void run() {
			taskId = Thread.currentThread().getName();
			LOG.log("Start Task "+taskId);
			while (runningFlag.get() && !Thread.currentThread().isInterrupted()) {
				PartitionReader preader = null;
				try {
					cache.clear();
					// check whether to release hold partitions
					checkRelease();
					// get on  a readable reader, if no readable reader exist , block here.
					preader=borrowReadableReaderAndBlokingIfNotExist();
					//check zk down. TODO: consider it carefully whether we really need it.
					// read a batch data <events>.
					cache = preader.readEvents();
					// return back this reader and all holding unreadable reader to <queue> for other threads.
					if(cache.size()<=0){
						hold(preader);
						preader=null;
					}else{
						preader.clearIdleStats();
						// release all the hold partitions
						while (!hold.isEmpty()) {
							PartitionReader p = hold.poll();
							putPartitionToQueue(p);
						}
					}
					//enable events visible for processing.
					stream.sink(cache);
				}catch(KafkaPartitionReaderException e){