public class TaskContent {
	private String group;
	private String topic;
	private Integer diapause; //
	private Integer threshold;
	private String mailTo;

	public TaskContent() {
		super();
	}

	public TaskContent(String group, String topic, Integer diapause, Integer threshold, String mailTo) {
		super();
		this.group = group;
		this.topic = topic;
		this.diapause = diapause;
		this.threshold = threshold;
		this.mailTo = mailTo;
	}

	public Integer getDiapause() {
		return diapause;
	}

	public void setDiapause(Integer diapause) {
		this.diapause = diapause;
	}

	public Integer getThreshold() {
		return threshold;
	}

	public void setThreshold(Integer threshold) {
		this.threshold = threshold;
	}

	public String getMailTo() {
		return mailTo;
	}

	public void setMailTo(String mailTo) {
		this.mailTo = mailTo;
	}

	public String getGroup() {
		return group;
	}

	public void setGroup(String group) {
		this.group = group;
	}

	public String getTopic() {
		return topic;
	}

	public void setTopic(String topic) {
		this.topic = topic;
	}

	@Override
	public int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result + ((group == null) ? 0 : group.hashCode());
		result = prime * result + ((topic == null) ? 0 : topic.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;
		TaskContent other = (TaskContent) obj;
		if (group == null) {
			if (other.group != null)
				return false;
		} else if (!group.equals(other.group))
			return false;
		if (topic == null) {
			if (other.topic != null)
				return false;
		} else if (!topic.equals(other.topic))
			return false;
		return true;
	}

	@Override
	public String toString() {
		return "TaskContent [group=" + group + ", topic=" + topic + ", diapause=" + diapause + ", threshold="
				+ threshold + ", mailTo=" + mailTo + "]";
	}

} public class TaskHandler implements Runnable {
	private static Logger LOG = LoggerFactory.getLogger(TaskHandler.class);

	@Override
	public void run() {
		try {
			while (true) {
				KafkaInfo kafkaInfo = SystemManager.offsetInfoCacheQueue.take();

				Map<String, Set<OffsetInfo>> cachedTriggeredOffsetInfo = new HashMap<String, Set<OffsetInfo>>();

				List<OffsetInfo> offsetInfoList = kafkaInfo.getOffsets();
				TaskContent task = null;
				Set<TaskContent> tasks = new HashSet<TaskContent>();
				for (OffsetInfo offsetInfo : offsetInfoList) {
					task = TaskManager.getTask(offsetInfo.getGroup(), offsetInfo.getTopic());
					if (task == null) {
						continue;
					}
					tasks.add(task);
					if (offsetInfo.getLag() > task.getThreshold()) {
						if (cachedTriggeredOffsetInfo.containsKey(offsetInfo.getTopic())) {
							cachedTriggeredOffsetInfo.get(offsetInfo.getTopic()).add(offsetInfo);
						} else {
							Set<OffsetInfo> offsetInfos = new HashSet<OffsetInfo>();
							offsetInfos.add(offsetInfo);
							cachedTriggeredOffsetInfo.put(offsetInfo.getTopic(), offsetInfos);
						}
					}
				}
				for (TaskContent _task : tasks) {
					Long lastSendTime = TaskManager.cachedLastSendTime.get(_task.getGroup() + "_" + _task.getTopic());
					// if alerting task keeping triggered, that make sure that
					// we do not keep on receiving email in a short time
					if (lastSendTime != null) {
						if ((System.currentTimeMillis() - lastSendTime) > _task.getDiapause() * 60 * 1000) {
							generateEmailContentAndSend(_task, cachedTriggeredOffsetInfo.get(_task.getTopic()));
						}
					} else {
						TaskManager.cachedLastSendTime.put(_task.getGroup() + "_" + _task.getTopic(),
								System.currentTimeMillis());
						generateEmailContentAndSend(_task, cachedTriggeredOffsetInfo.get(_task.getTopic()));
					}
				}
			}
		} catch (InterruptedException e) {
			LOG.error("Something went wrong when handle alert task...", e);
		}
	}

	private void generateEmailContentAndSend(TaskContent _task, Set<OffsetInfo> offsetInfos) {
		TaskManager.cachedLastSendTime.put(_task.getGroup() + "_" + _task.getTopic(), System.currentTimeMillis());
		Template template = new Template();
		StringBuilder blabla;
		blabla = new StringBuilder();
		for (OffsetInfo offsetInfo : offsetInfos) {
			blabla.append("<tr><td style=\"border: 1px solid #ddd;\">" + offsetInfo.getGroup() + "</td>");
			blabla.append("<td style=\"border: 1px solid #ddd;\">" + offsetInfo.getTopic() + "</td>");
			blabla.append("<td style=\"border: 1px solid #ddd;\">" + offsetInfo.getPartition() + "</td>");
			blabla.append("<td style=\"border: 1px solid #ddd;\">" + offsetInfo.getLag() + "</td><tr>");
		}
		template.insertTr(blabla.toString());
		EmailSender.sendEmail(template.getContent(), _task.getMailTo(), _task.getGroup() + "_" + _task.getTopic());
	}

} public class TaskManager {
	protected static final Logger logger = LoggerFactory.getLogger(TaskManager.class);

	public static String taskFolder;

	public static Map<String, Long> cachedLastSendTime = new HashMap<String, Long>();

	// k:group v:(k:topic v:taskContent)
	private static Map<String, Map<String, TaskContent>> tasks;

	public static void init(AppConfig config) {
		taskFolder = config.getTaskFolder();
		tasks = Collections.synchronizedMap(new HashMap<String, Map<String, TaskContent>>());
	}

	public static TaskContent getTask(String group, String topic) {
		if (tasks.containsKey(group)) {
			return tasks.get(group).get(topic);
		}
		return null;
	}

	public static boolean exits(TaskContent taskContent) {
		if (tasks.containsKey(taskContent.getGroup())) {
			return tasks.get(taskContent.getGroup()).containsKey(taskContent.getTopic());
		}
		return false;
	}

	public static void addTask(TaskContent taskContent) throws JSONException {
		Map<String, TaskContent> task = null;
		if (tasks.containsKey(taskContent.getGroup())) {
			task = tasks.get(taskContent.getGroup());
		} else {
			task = new HashMap<String, TaskContent>();
		}
		task.put(taskContent.getTopic(), taskContent);
		tasks.put(taskContent.getGroup(), task);
	}

	public static void deleteTask(String taskNameToRemove) {
		String[] group_topic = taskNameToRemove.split("-");
		if (exits(new TaskContent(group_topic[0], group_topic[1], null, null, null))) {
			TaskContent taskToDelete = tasks.get(group_topic[0]).remove(group_topic[1]);
			deleteTaskFile(taskToDelete);
		}
	}

	public static void deleteTaskFile(TaskContent task) {
		new File(taskFolder + "/" + task.getGroup() + "-" + task.getTopic() + ".task").delete();
	}

	public static Set<TaskContent> getTasks() {
		Set<TaskContent> _tasks = new HashSet<TaskContent>();
		tasks.forEach((group, topic_task) -> {
			Set<Entry<String, TaskContent>> entrySet = topic_task.entrySet();
			entrySet.forEach(entry -> {
				_tasks.add(entry.getValue());
			});
		});
		return _tasks;
	}

	public static void saveTaskToFileAndAddToTasks(TaskContent taskContent) {
		try {
			PrintWriter writer = new PrintWriter(
					taskFolder + "/" + taskContent.getGroup() + "-" + taskContent.getTopic() + ".task", "UTF-8");
			writer.println(new Gson().toJson(taskContent));
			writer.close();
			addTask(taskContent);
		} catch (IOException e) {
			logger.error("create task file failed!", e);
		}
	}
} public class WorkerThreadFactory implements ThreadFactory {
	private AtomicInteger counter = new AtomicInteger(0);
	private String prefix = "";

	public WorkerThreadFactory(String prefix) {
		this.prefix = prefix;
	}

	@Override
	public Thread newThread(Runnable r) {
		Thread t = new Thread(r, prefix + "-" + counter.incrementAndGet());
		t.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {

			@Override
			public void uncaughtException(Thread t, Throwable e) {
				LoggerFactory.getLogger(t.getName()).error(e.getMessage(), e);
			}
		});
		return t;
	}

} public class AppConfig {
	private String clusterName;
	private String apiType = "";
	private String esHosts;
	private String esIndex = "";
	private Integer dataCollectFrequency = 1;

	private String zkHosts;
	private Integer zkSessionTimeout = 30 * 1000;
	private Integer zkConnectionTimeout = 30 * 1000;

	private Boolean isAlertEnabled = false;

	private Integer offsetInfoCacheQueue = 500;
	private Integer offsetInfoHandler;
	private String taskFolder = "tasks";

	private Boolean smtpAuth = false;
	private String smtpUser;
	private String smtpPasswd;
	private String smtpServer;
	private String mailSender;
	private String mailSubject;

	private Long excludeByLastSeen = 2592000L;

	public String getApiType() {
		return apiType;
	}

	public void setApiType(String apiType) {
		this.apiType = apiType;
	}

	public String getZkHosts() {
		return zkHosts;
	}

	public void setZkHosts(String zkHosts) {
		this.zkHosts = zkHosts;
	}

	public Boolean getIsAlertEnabled() {
		return isAlertEnabled;
	}

	public void setIsAlertEnabled(Boolean isAlertEnabled) {
		this.isAlertEnabled = isAlertEnabled;
	}

	public String getEsHosts() {
		return esHosts;
	}

	public void setEsHosts(String esHosts) {
		this.esHosts = esHosts;
	}

	public String getEsIndex() {
		return esIndex.toLowerCase();
	}

	public void setEsIndex(String esIndex) {
		this.esIndex = esIndex;
	}

	public Integer getDataCollectFrequency() {
		return dataCollectFrequency;
	}

	public void setDataCollectFrequency(Integer dataCollectFrequency) {
		this.dataCollectFrequency = dataCollectFrequency;
	}

	public Integer getZkSessionTimeout() {
		return zkSessionTimeout;
	}

	public void setZkSessionTimeout(Integer zkSessionTimeout) {
		this.zkSessionTimeout = zkSessionTimeout;
	}

	public Integer getZkConnectionTimeout() {
		return zkConnectionTimeout;
	}

	public void setZkConnectionTimeout(Integer zkConnectionTimeout) {
		this.zkConnectionTimeout = zkConnectionTimeout;
	}

	public Integer getOffsetInfoCacheQueue() {
		return offsetInfoCacheQueue;
	}

	public void setOffsetInfoCacheQueue(Integer offsetInfoCacheQueue) {
		this.offsetInfoCacheQueue = offsetInfoCacheQueue;
	}

	public Integer getOffsetInfoHandler() {
		return offsetInfoHandler;
	}

	public void setOffsetInfoHandler(Integer offsetInfoHandler) {
		this.offsetInfoHandler = offsetInfoHandler;
	}

	public Boolean getSmtpAuth() {
		return smtpAuth;
	}

	public void setSmtpAuth(Boolean smtpAuth) {
		this.smtpAuth = smtpAuth;
	}

	public String getSmtpUser() {
		return smtpUser;
	}

	public void setSmtpUser(String smtpUser) {
		this.smtpUser = smtpUser;
	}

	public String getSmtpPasswd() {
		return smtpPasswd;
	}

	public void setSmtpPasswd(String smtpPasswd) {
		this.smtpPasswd = smtpPasswd;
	}

	public String getSmtpServer() {
		return smtpServer;
	}

	public void setSmtpServer(String smtpServer) {
		this.smtpServer = smtpServer;
	}

	public String getMailSender() {
		return mailSender;
	}

	public void setMailSender(String mailSender) {
		this.mailSender = mailSender;
	}

	public String getMailSubject() {
		return mailSubject;
	}

	public void setMailSubject(String mailSubject) {
		this.mailSubject = mailSubject;
	}

	public String getTaskFolder() {
		return taskFolder;
	}

	public void setTaskFolder(String taskFolder) {
		this.taskFolder = taskFolder;
	}

	public Long getExcludeByLastSeen() {
		return excludeByLastSeen;
	}

	public void setExcludeByLastSeen(Long excludeByLastSeen) {
		this.excludeByLastSeen = excludeByLastSeen;
	}

	public void setClusterName(String clusterName){
		this.clusterName = clusterName;
	}
	public String getClusterName(){
		if(this.clusterName == null){
			return "";
		}
		return this.clusterName;
	}
} public abstract class OffsetGetter {

	static Logger LOG = LoggerFactory.getLogger(OffsetGetter.class);

	Map<Integer, SimpleConsumer> consumerMap = new HashMap<Integer, SimpleConsumer>();

	protected static boolean earliest = false;

	protected static final String clientId = "Kmanager";

	/**
	 * <code>
	 * 			{
				    "listener_security_protocol_map": {
				        "PLAINTEXT": "PLAINTEXT"
				    },
				    "endpoints": [
				        "PLAINTEXT://10.16.238.94:8092"
				    ],
				    "jmx_port": 8888,
				    "host": "10.16.238.94",
				    "timestamp": "1497505727915",
				    "port": 8092,
				    "version": 4
				}
	 * </code>
	 * 
	 * @param bId
	 * @return
	 */
	public SimpleConsumer getConsumer(int bId) {
		String jsonString = null;
		JSONObject brokerInfoJson = null;
		try {
			Tuple2<Option<String>, Stat> brokerInfo_stat = ZKUtils.getZKUtilsFromKafka()
					.readDataMaybeNull(ZkUtils.BrokerIdsPath() + "/" + bId);
			if (brokerInfo_stat._1.isEmpty()) {
				throw new BrokerNotAvailableException("Broker id " + bId + " does not exist");
			}
			jsonString = brokerInfo_stat._1.get();
			brokerInfoJson = new JSONObject(jsonString);
		} catch (Exception e) {
			throw new BrokerNotAvailableException("Broker id:" + bId + " " + jsonString);
		}
		return new SimpleConsumer(brokerInfoJson.getString("host"), brokerInfoJson.getInt("port"), 10000, 100000,
				"ConsumerOffsetChecker");
	}

	public List<OffsetInfo> processTopic(String group, String topic) throws Exception {
		List<String> partitionIds = null;
		try {
			partitionIds = JavaConversions.seqAsJavaList(ZKUtils.getZKUtilsFromKafka()
					.getChildren(ZkUtils.BrokerTopicsPath() + "/" + topic + "/partitions"));
		} catch (Exception e) {
			if (e instanceof NoNodeException) {
				LOG.warn("Is topic >" + topic + "< exists!", e);
				return null;
			}
		}
		List<OffsetInfo> offsetInfos = new ArrayList<OffsetInfo>();
		OffsetInfo offsetInfo = null;
		if (partitionIds == null) {
			// TODO that topic exists in consumer node but not in topics node?!
			return null;
		}

		for (String partitionId : partitionIds) {
			offsetInfo = processPartition(group, topic, partitionId);
			if (offsetInfo != null) {
				offsetInfos.add(offsetInfo);
			}
		}
		return offsetInfos;
	}

	public List<BrokerInfo> brokerInfo() {
		List<BrokerInfo> binfos = new ArrayList<BrokerInfo>();
		Set<Integer> bids = consumerMap.keySet();
		BrokerInfo binfo = null;
		SimpleConsumer consumer = null;
		for (int bid : bids) {
			consumer = consumerMap.get(bid);
			binfo = new BrokerInfo(bid, consumer.host(), consumer.port());
			binfos.add(binfo);
		}
		return binfos;

	}

	public List<OffsetInfo> offsetInfo(String group, List<String> topics) throws Exception {
		List<OffsetInfo> offsetInfos = new ArrayList<OffsetInfo>();
		if (topics.isEmpty()) {
			topics = getTopicList(group);
		}
		for (String topic : topics) {
			List<OffsetInfo> offsetInfoList = processTopic(group, topic);
			if (!(offsetInfoList == null))
				offsetInfos.addAll(offsetInfoList);
		}
		return offsetInfos;
	}

	// get information about a consumer group and the topics it consumes
	public KafkaInfo getInfo(String group, List<String> topics) throws Exception {
		List<OffsetInfo> offsetInfos = offsetInfo(group, topics);
		Collections.sort(offsetInfos, new Comparator<OffsetInfo>() {

			@Override
			public int compare(OffsetInfo o1, OffsetInfo o2) {
				int flag = o1.getTopic().compareTo(o2.getTopic());
				if (flag == 0) {
					return o1.getPartition().compareTo(o2.getPartition());
				} else {
					return flag;
				}
			}

		});
//		return new KafkaInfo(group, brokerInfo(), offsetInfos);
		return new KafkaInfo(group, null, offsetInfos);
	}

	public List<String> getTopics() {
		List<String> topics = null;
		try {
			topics = JavaConversions
					.seqAsJavaList(ZKUtils.getZKUtilsFromKafka().getChildren(ZkUtils.BrokerTopicsPath()));
		} catch (Exception e) {
			LOG.error("could not get topics because of " + e.getMessage(), e);
		}
		return topics;
	}

	public Node getClusterViz() {
		Node rootNode = new Node("KafkaCluster");
		List<Node> childNodes = new ArrayList<Node>();
		List<Broker> brokers = JavaConversions.seqAsJavaList(ZKUtils.getZKUtilsFromKafka().getAllBrokersInCluster());
		brokers.forEach(broker -> {
			List<EndPoint> endPoints = JavaConversions.seqAsJavaList(broker.endPoints().seq());
			childNodes.add(new Node(broker.id() + ":" + endPoints.get(0).host() + ":" + endPoints.get(0).port(), null));
		});
		rootNode.setChildren(childNodes);
		return rootNode;
	}

	/**
	 * Returns details for a given topic such as the consumers pulling off of it
	 * 
	 * @param topic
	 * @return
	 */
	public TopicDetails getTopicDetail(String topic) {
		Map<String, List<String>> topicMap = getActiveTopicMap();
		List<String> consumers = null;
		List<ConsumerDetail> consumerDetails = new ArrayList<ConsumerDetail>();
		if (topicMap.containsKey(topic)) {
			consumers = topicMap.get(topic);
			for (String consumer : consumers) {
				consumerDetails.add(new ConsumerDetail(consumer));
			}
		}
		return new TopicDetails(consumerDetails);
	}

	public List<ConsumerDetail> mapConsumerDetails(List<String> consumers) {
		List<ConsumerDetail> consumerDetails = new ArrayList<ConsumerDetail>();
		for (String consumer : consumers) {
			consumerDetails.add(new ConsumerDetail(consumer));
		}
		return consumerDetails;
	}

	/**
	 * Returns details for a given topic such as the active consumers pulling off of
	 * it and for each of the active consumers it will return the consumer data
	 * 
	 * @throws Exception
	 */
	public TopicAndConsumersDetails getTopicAndConsumersDetail(String topic) throws Exception {
		Map<String, List<String>> topicMap = getTopicMap();
		Map<String, List<String>> activeTopicMap = getActiveTopicMap();

		List<KafkaInfo> activeConsumers = new ArrayList<KafkaInfo>();
		if (activeTopicMap.containsKey(topic)) {
			activeConsumers = mapConsumersToKafkaInfo(activeTopicMap.get(topic), topic);
		}

		List<KafkaInfo> inActiveConsumers = new ArrayList<KafkaInfo>();
		if (!activeTopicMap.containsKey(topic) && topicMap.containsKey(topic)) {
			inActiveConsumers = mapConsumersToKafkaInfo(topicMap.get(topic), topic);
		}
		return new TopicAndConsumersDetails(activeConsumers, inActiveConsumers);
	}

	public List<String> getActiveConsumer(String topic) throws Exception {
		Map<String, List<String>> activeTopicMap = getActiveTopicMap();

		List<String> activeConsumers = null;
		if (activeTopicMap.containsKey(topic)) {
			activeConsumers = activeTopicMap.get(topic);
		}
		return activeConsumers;
	}

	private List<KafkaInfo> mapConsumersToKafkaInfo(List<String> consumers, String topic) throws Exception {
		List<KafkaInfo> kafkaInfos = new ArrayList<KafkaInfo>();
		List<String> topics = new ArrayList<String>();
		topics.add(topic);
		for (String consumer : consumers) {
			kafkaInfos.add(getInfo(consumer, topics));
		}
		return kafkaInfos;
	}

	public Node getActiveTopics() {
		Map<String, List<String>> activeTopicMap = getActiveTopicMap();
		List<Node> activeTopics = new ArrayList<Node>();
		List<Node> temp = null;
		Set<String> keys = activeTopicMap.keySet();
		for (String topic : keys) {
			List<String> consumers = activeTopicMap.get(topic);
			temp = new ArrayList<Node>();
			for (String consumer : consumers) {
				temp.add(new Node(consumer));
			}
			activeTopics.add(new Node(topic, temp));
		}
		return new Node("ActiveTopics", activeTopics);
	}

	public void close() {
		try {
			Iterator<Entry<Integer, SimpleConsumer>> it = consumerMap.entrySet().iterator();
			while (it.hasNext()) {
				LOG.debug("Closing consumer: " + it.next().getValue().clientId());
				it.next().getValue().close();
			}
		} catch (Exception e) {
			LOG.warn("Close SimpleConsumer: " + e.getMessage());
		}
	}

	public abstract Map<String, List<String>> getTopicMap();

	public abstract Map<String, List<String>> getActiveTopicMap();

	public abstract List<String> getTopicList(String group);

	public abstract OffsetInfo processPartition(String group, String topic, String partitionId);

	public abstract List<String> getGroups();

} public class ZKOffsetGetter extends OffsetGetter {

	private static Long excludeByLastSeen = 604_800_000L;

	public ZKOffsetGetter(AppConfig config) {
		excludeByLastSeen = config.getExcludeByLastSeen() * 1000;
		ZKUtils.init(config.getZkHosts(), config.getZkSessionTimeout(), config.getZkConnectionTimeout());
	}

	@Override
	public List<String> getGroups() {
		return ZKUtils.getChildren(ZkUtils.ConsumersPath());
	}

	public List<String> checkIfTopicExistsAndRemoveThatNot(List<String> topicsProvide, String group) {
		List<String> checkedTopic = new ArrayList<String>();
		List<String> allTopicsBelongTheGroup = getTopicList(group);
		topicsProvide.forEach(topic -> {
			if (allTopicsBelongTheGroup.contains(topic)) {
				checkedTopic.add(topic);
			}
		});
		return checkedTopic;
	}

	@Override
	public Map<String, List<String>> getTopicMap() {
		Map<String, List<String>> topicGroupsMap = new HashMap<String, List<String>>();
		List<String> groups = getGroups();
		List<String> topics = null;
		for (String group : groups) {
			topics = getTopicList(group);
			topics.forEach(topic -> {
				List<String> _groups = null;
				if (topicGroupsMap.containsKey(topic)) {
					_groups = topicGroupsMap.get(topic);
					_groups.add(group);
				} else {
					_groups = new ArrayList<String>();
					_groups.add(group);
				}
				topicGroupsMap.put(topic, _groups);
			});
		}
		return topicGroupsMap;
	}

	@Override
	public Map<String, List<String>> getActiveTopicMap() {
		Map<String, List<String>> topicGroupsMap = new HashMap<String, List<String>>();
		List<String> consumers = ZKUtils.getChildren(ZkUtils.ConsumersPath());
		for (String consumer : consumers) {
			Map<String, scala.collection.immutable.List<ConsumerThreadId>> consumer_consumerThreadId = null;
			try {
				consumer_consumerThreadId = JavaConversions
						.mapAsJavaMap(ZKUtils.getZKUtilsFromKafka().getConsumersPerTopic(consumer, true));
			} catch (Exception e) {
				LOG.warn("getActiveTopicMap-> getConsumersPerTopic for group: " + consumer + "failed! "
						+ e.getMessage());
				// TODO /consumers/{group}/ids/{id} 节点的内容不符合要求。这个group有问题
				continue;
			}
			Set<String> topics = consumer_consumerThreadId.keySet();
			topics.forEach(topic -> {
				List<String> _groups = null;
				if (topicGroupsMap.containsKey(topic)) {
					_groups = topicGroupsMap.get(topic);
					_groups.add(consumer);
				} else {
					_groups = new ArrayList<String>();
					_groups.add(consumer);
				}
				topicGroupsMap.put(topic, _groups);
			});
		}
		return topicGroupsMap;
	}

	@Override
	public List<String> getTopicList(String group) {
		return ZKUtils.getChildren(ZkUtils.ConsumersPath() + '/' + group + "/offsets");
	}

	@Override
	public OffsetInfo processPartition(String group, String topic, String partitionId) {
		OffsetInfo offsetInfo = null;
		Tuple2<String, Stat> offset_stat = readZkData(
				ZkUtils.ConsumersPath() + "/" + group + "/" + "offsets/" + topic + "/" + partitionId);
		if (offset_stat == null) {
			return null;
		}
		if (System.currentTimeMillis() - offset_stat._2().getMtime() > excludeByLastSeen) {
			// TODO 对于最后一次消费时间为一周前的，直接抛弃。是否维护一个被排除的partition 列表？
			return null;
		}
		ZkDataAndStat dataAndStat = ZKUtils
				.readDataMaybeNull(ZkUtils.ConsumersPath() + "/" + group + "/" + "owners/" + topic + "/" + partitionId);
		try {
			Integer leader = (Integer) ZKUtils.getZKUtilsFromKafka()
					.getLeaderForPartition(topic, Integer.parseInt(partitionId)).get();

			SimpleConsumer consumer = null;
			if (consumerMap.containsKey(leader)) {
				consumer = consumerMap.get(leader);
			} else {
				consumer = getConsumer(leader);
				consumerMap.put(leader, consumer);
			}

			TopicAndPartition topicAndPartition = new TopicAndPartition(topic, Integer.parseInt(partitionId));

			Map<TopicAndPartition, PartitionOffsetRequestInfo> tpMap = new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
			tpMap.put(topicAndPartition, new PartitionOffsetRequestInfo(
					earliest ? kafka.api.OffsetRequest.EarliestTime() : kafka.api.OffsetRequest.LatestTime(), 1));
			OffsetRequest request = new OffsetRequest(tpMap, kafka.api.OffsetRequest.CurrentVersion(),
					String.format("%s_%s_%s", topic, partitionId, clientId));
			OffsetResponse response = null;
			response = consumer.getOffsetsBefore(request);
			if (response.hasError()) {
				LOG.error("error fetching data Offset from the Broker {}. reason: {}", leader,
						response.errorCode(topic, Integer.parseInt(partitionId)));
				throw new RuntimeException("fetching offset error!");
			}
			long[] offsets = response.offsets(topic, Integer.parseInt(partitionId));

			if (dataAndStat.getData() == null) { // Owner not available
				// TODO dataAndStat that partition may not have any owner
				offsetInfo = new OffsetInfo(group, topic, Integer.parseInt(partitionId),
						Long.parseLong(offset_stat._1()), offsets[offsets.length - 1], "NA",
						offset_stat._2().getCtime(), offset_stat._2().getMtime());
			} else {
				offsetInfo = new OffsetInfo(group, topic, Integer.parseInt(partitionId),
						Long.parseLong(offset_stat._1()), offsets[offsets.length - 1], dataAndStat.getData(),
						offset_stat._2().getCtime(), offset_stat._2().getMtime());
			}
		} catch (Exception e) {

			if (e instanceof ZkNoNodeException) {

			} else if (e instanceof NoNodeException) {
				// TODO dataAndStat that partition may not have any owner

			} else if (e instanceof BrokerNotAvailableException) {
				// TODO broker id -1 ? Alerting???
				LOG.warn(String.format("Get leader partition for [group: %s, topic: %s, partition: %s] faild!", group,
						topic, partitionId), e.getMessage());
				if (dataAndStat.getData() == null) { // Owner not available
					// TODO dataAndStat that partition may not have any owner
					offsetInfo = new OffsetInfo(group, topic, Integer.parseInt(partitionId),
							Long.parseLong(offset_stat._1()), -1l, "NA", offset_stat._2().getCtime(),
							offset_stat._2().getMtime());
				} else {
					offsetInfo = new OffsetInfo(group, topic, Integer.parseInt(partitionId),
							Long.parseLong(offset_stat._1()), -1l, dataAndStat.getData(), offset_stat._2().getCtime(),
							offset_stat._2().getMtime());
				}
			} else {
				throw new RuntimeException("Something went wrong!" + e);
			}
		}
		return offsetInfo;
	}

	private Tuple2<String, Stat> readZkData(String path) {
		Tuple2<String, Stat> offset_stat = null;
		try {
			offset_stat = ZKUtils.getZKUtilsFromKafka().readData(path);
		} catch (ZkNoNodeException znne) {
			// TODO no offset record in zk?
		}
		return offset_stat;
	}

	@Override
	public void close() {
		ZKUtils.close();
		super.close();
	}

} public class ElasticsearchOffsetDB implements OffsetDB<Ielasticsearch> {

  private Ielasticsearch esUtil;
  private String indexPrefix;
  private String docType;

  private static final SimpleDateFormat sFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ");

  public ElasticsearchOffsetDB(AppConfig config) {
    if (config.getApiType().equalsIgnoreCase("Java API")) {
      esUtil = new ElasticsearchJavaUtil(config.getEsHosts());
    } else {
      esUtil = new ElasticsearchRESTUtil(config.getEsHosts());
    }
    setIndexAndType(config.getEsIndex(), SystemManager.getElasticSearchOffsetType());
  }

  public void setIndexAndType(String index, String docType) {
    this.indexPrefix = index + "-";
    this.docType = docType;
  }

  /*
   * (non-Javadoc)
   * 
   * @see com.chickling.kmanager.core.db.OffsetDb#insert(long,
   * com.chickling.kmanager.model.OffsetInfo)
   */
  @Override
  public void insert(long timestamp, OffsetInfo offsetInfo) {
    // TODO Auto-generated method stub
  }

  /*
   * (non-Javadoc)
   * 
   * @see com.chickling.kmanager.core.db.OffsetDb#batchInsert(java. util.List)
   */
  @Override
  public void batchInsert(List<OffsetInfo> offsetInfoList) {
    long now = System.currentTimeMillis();
    Calendar cal = Calendar.getInstance();
    cal.setTimeInMillis(now);
    cal.set(Calendar.MILLISECOND, 0);
    cal.set(Calendar.SECOND, 0);

    JSONObject data = new JSONObject();
    for (int i = 0; i < offsetInfoList.size(); i++) {
      data.put(i + "", generateRecord(cal.getTimeInMillis(), offsetInfoList.get(i)));
    }
    esUtil.bulkIndex(data, docType, indexPrefix);
  }

  private JSONObject generateRecord(long timestamp, OffsetInfo offsetInfo) {
    JSONObject data = new JSONObject();
    data.put("group", offsetInfo.getGroup());
    data.put("topic", offsetInfo.getTopic());
    data.put("partition", offsetInfo.getPartition());
    data.put("offset", offsetInfo.getOffset());
    data.put("logSize", offsetInfo.getLogSize());
    data.put("owner", offsetInfo.getOwner()/* owner.getOrElse("NA") */);
    data.put("date", sFormat.format(new Date(timestamp)));
    data.put("timestamp", timestamp);
    data.put("creation", offsetInfo.getCreation());
    data.put("modified", offsetInfo.getModified());
    data.put("lag", offsetInfo.getLag());
    return data;
  }

  /*
   * (non-Javadoc)
   * 
   * @see com.chickling.kmanager.core.db.OffsetDb#offsetHistory()
   */
  @Override
  public OffsetHistory offsetHistory(String group, String topic) {
    List<OffsetPoints> offsetPointsList = esUtil.offsetHistory(indexPrefix, docType, group, topic);
    CommonUtils.sortByTimestampThenPartition(offsetPointsList);
    return new OffsetHistory(group, topic, offsetPointsList);
  }

  @Override
  public OffsetHistory offsetHistory(OffsetHistoryQueryParams params) {
    List<OffsetPoints> offsetPointsList = esUtil.scrollsSearcher(params, docType, indexPrefix);
    CommonUtils.sortByTimestampThenPartition(offsetPointsList);
    return new OffsetHistory(params.getGroup(), params.getTopic(), offsetPointsList);
  }

  @Override
  public boolean check() {
    return esUtil.check();
  }

  @Override
  public void close() {
    esUtil.close();
  }

  @Override
  public Ielasticsearch getDB() {
    return esUtil;
  }

} public interface OffsetDB<T> {
  T getDB();

  void insert(long timestamp, OffsetInfo offsetInfo);

  void batchInsert(List<OffsetInfo> offsetInfoList);

  OffsetHistory offsetHistory(String group, String topic);

  OffsetHistory offsetHistory(OffsetHistoryQueryParams params);

  void close();

  boolean check();
} public class EmailSender {
  private static Logger LOG = LoggerFactory.getLogger(EmailSender.class);

  private static AppConfig config;

  public static void setConfig(AppConfig _config) {
    config = _config;
  }

  public static void sendEmail(String message, String sendTo, String group_topic) {
    Properties properties = System.getProperties();

    if (config.getSmtpAuth()) {
      properties.setProperty("mail.user", config.getSmtpUser());
      properties.setProperty("mail.password", config.getSmtpPasswd());
    }
    properties.setProperty("mail.smtp.host", config.getSmtpServer());

    Session session = Session.getDefaultInstance(properties);

    MimeMessage mimeMessage = new MimeMessage(session);

    try {
      String[] sendToArr = sendTo.split(";");
      mimeMessage.setFrom(new InternetAddress(config.getMailSender()));
      if (sendToArr.length > 1) {
        String cc = "";
        for (int i = 1; i < sendToArr.length; i++) {
          cc += i == sendToArr.length - 1 ? sendToArr[i] : sendToArr[i] + ",";
        }
        mimeMessage.addRecipients(Message.RecipientType.CC, InternetAddress.parse(cc));
      }
      mimeMessage.addRecipient(Message.RecipientType.TO, new InternetAddress(sendToArr[0]));

      String[] group_topicArr = group_topic.split("_");
      String subject = config.getMailSubject();
      if (subject.contains("{group}")) {
        subject = subject.replace("{group}", group_topicArr[0]);
      }
      if (subject.contains("{topic}")) {
        subject = subject.replace("{topic}", group_topicArr[1]);
      }
      mimeMessage.setSubject(subject);
      mimeMessage.setSentDate(new Date());
      mimeMessage.setContent(message, "text/html");

      Transport.send(mimeMessage);
    } catch (

    Exception e) {
      LOG.error("sendEmail faild!", e);
    }
  }
} public class FormatedMeterMetric {
  private Long count;
  private String meanRate;
  private String oneMinuteRate;
  private String fiveMinuteRate;
  private String fifteenMinuteRate;

  public FormatedMeterMetric() {
    super();
  }

  public FormatedMeterMetric(MeterMetric metric) {
    this(metric.getCount(), MetricUtils.sizeFormat(metric.getMeanRate()), MetricUtils.sizeFormat(metric.getOneMinuteRate()),
        MetricUtils.sizeFormat(metric.getFiveMinuteRate()), MetricUtils.sizeFormat(metric.getFifteenMinuteRate()));
  }

  public FormatedMeterMetric(MeterMetric metric, int interation) {
    this(metric.getCount(), MetricUtils.rateFormat(metric.getMeanRate(), interation),
        MetricUtils.rateFormat(metric.getOneMinuteRate(), interation), MetricUtils.rateFormat(metric.getFiveMinuteRate(), interation),
        MetricUtils.rateFormat(metric.getFifteenMinuteRate(), interation));
  }

  public FormatedMeterMetric(Long count, String meanRate, String oneMinuteRate, String fiveMinuteRate, String fifteenMinuteRate) {
    super();
    this.count = count;
    this.meanRate = meanRate;
    this.oneMinuteRate = oneMinuteRate;
    this.fiveMinuteRate = fiveMinuteRate;
    this.fifteenMinuteRate = fifteenMinuteRate;
  }

  public Long getCount() {
    return count;
  }

  public void setCount(Long count) {
    this.count = count;
  }

  public String getFifteenMinuteRate() {
    return fifteenMinuteRate;
  }

  public void setFifteenMinuteRate(String fifteenMinuteRate) {
    this.fifteenMinuteRate = fifteenMinuteRate;
  }

  public String getFiveMinuteRate() {
    return fiveMinuteRate;
  }

  public void setFiveMinuteRate(String fiveMinuteRate) {
    this.fiveMinuteRate = fiveMinuteRate;
  }

  public String getOneMinuteRate() {
    return oneMinuteRate;
  }

  public void setOneMinuteRate(String oneMinuteRate) {
    this.oneMinuteRate = oneMinuteRate;
  }

  public String getMeanRate() {
    return meanRate;
  }

  public void setMeanRate(String meanRate) {
    this.meanRate = meanRate;
  }

} public interface JMXExecutor {

	public void doWithConnection(MBeanServerConnection mBeanServerConnection);

} public class KafkaJMX {
	private static Logger LOG = LoggerFactory.getLogger(KafkaJMX.class);

	private Map<String, Object> defaultJmxConnectorProperties = new HashMap<String, Object>();

	public KafkaJMX() {
		initDefaultJMXConnectorProperties();
	}

	private void initDefaultJMXConnectorProperties() {
		defaultJmxConnectorProperties.put("jmx.remote.x.request.waiting.timeout", "3000");
		defaultJmxConnectorProperties.put("jmx.remote.x.notification.fetch.timeout", "3000");
		defaultJmxConnectorProperties.put("sun.rmi.transport.connectionTimeout", "3000");
		defaultJmxConnectorProperties.put("sun.rmi.transport.tcp.handshakeTimeout", "3000");
		defaultJmxConnectorProperties.put("sun.rmi.transport.tcp.responseTimeout", "3000");
	}

	public void doWithConnection(String jmxHost, int jmxPort, Optional<String> jmxUser, Optional<String> jmxPasswd,
			boolean jmxSSL, JMXExecutor excutor) {
		String urlStr = "service:jmx:rmi:///jndi/rmi://" + jmxHost + ":" + jmxPort + "/jmxrmi";
		JMXConnector jmxc = null;
		try {
			JMXServiceURL url = new JMXServiceURL(urlStr);
			// authenticate
			Map<String, Object> env = new HashMap<String, Object>();
			String[] credentials = { jmxUser.orElse(""), jmxPasswd.orElse("") };
			env.put(JMXConnector.CREDENTIALS, credentials);

			if (jmxSSL) { // com.sun.management.jmxremote.registry.ssl=true
				env.put("com.sun.jndi.rmi.factory.socket", new SslRMIClientSocketFactory());
			}
			jmxc = JMXConnectorFactory.connect(url, env);
			excutor.doWithConnection(jmxc.getMBeanServerConnection());
		} catch (Exception e) {
			LOG.error("KafkaJMX doWithConnection error! " + e.getMessage());
		} finally {
			try {
				if (jmxc != null) {
					jmxc.close();
				}
			} catch (IOException e) {
				LOG.error("Close JMXConnector error! " + e.getMessage());
			}
		}
	}

} public class KafkaMetrics {
  private static Logger LOG = LoggerFactory.getLogger(KafkaMetrics.class);

  // kafka.network:type=RequestMetrics,name=RequestsPerSec,request={Produce|FetchConsumer|FetchFollower}

  // java.lang
  public static final String OPRATING_SYSTEM = "java.lang:type=OperatingSystem";
  public static final String RUNTIME = "java.lang:type=Runtime";

  // kafka.cluster
  public static final String IN_SYNC_REPLICAS_COUNT = "kafka.cluster:type=Partition,name=InSyncReplicasCount,topic={},partition={}";
  public static final String REPLICAS_COUNT = "kafka.cluster:type=Partition,name=ReplicasCount,topic={},partition={}";
  public static final String UNDER_REPLICATED = "kafka.cluster:type=Partition,name=UnderReplicated,topic={},partition={}";

  // kafka.controller
  public static final String LEADER_ELECTION_RATE_AND_TIME_MS = "kafka.controller:type=ControllerStats,name=LeaderElectionRateAndTimeMs";
  public static final String UNCLEAN_LEADER_ELECTIONS_PER_SEC = "kafka.controller:type=ControllerStats,name=UncleanLeaderElectionsPerSec";
  public static final String ACTIVE_CONTROLLER_COUNT = "kafka.controller:type=KafkaController,name=ActiveControllerCount";
  public static final String OFFLINE_PARTITIONS_COUNT = "kafka.controller:type=KafkaController,name=OfflinePartitionsCount";
  public static final String PREFERED_REPLICA_IMBALANCE = "kafka.controller:type=KafkaController,name=PreferredReplicaImbalanceCount";

  // kafka.log
  public static final String LOG_END_OFFSET = "kafka.log:type=Log,name=LogEndOffset,topic={},partition={}";
  public static final String LOG_START_OFFSET = "kafka.log:type=Log,name=LogStartOffset,topic={},partition={}";
  public static final String LOG_SEGMENTS_NUM = "kafka.log:type=Log,name=NumLogSegments,topic={},partition={}";
  public static final String LOG_SIZE = "kafka.log:type=Log,name=Size,topic={},partition={}";
  public static final String CLEANER_RECOPY_PERCENT = "kafka.log:type=LogCleaner,name=cleaner-recopy-percent";
  public static final String MAX_BUFFER_UTILIZATION_PERCENT = "kafka.log:type=LogCleaner,name=max-buffer-utilization-percent";
  public static final String MAX_CLEAN_TIME_SECS = "kafka.log:type=LogCleaner,name=max-clean-time-secs";
  public static final String MAX_DIRTY_PERCENT = "kafka.log:type=LogCleanerManager,name=max-dirty-percent";
  public static final String TIME_SINCE_LAST_RUN_CLEANER_MS = "kafka.log:type=LogCleanerManager,name=time-since-last-run-ms";

  // kafka.network
  public static final String PROCESSOR_IDLEPERCENT = "kafka.network:type=Processor,name=IdlePercent,networkProcessor={}";
  public static final String REQUEST_QUEUE_SIZE = "kafka.network:type=RequestChannel,name=RequestQueueSize";
  public static final String RESPONSE_QUEUE_SIZE = "kafka.network:type=RequestChannel,name=ResponseQueueSize";
  public static final String RESPONSE_QUEUE_SIZE_OF_PROCESSOR = "kafka.network:type=RequestChannel,name=ResponseQueueSize,processor={}";

  public static final String REQUEST_METRICS = "kafka.network:type=RequestMetrics,name={},request= {}";
  public static final String REQUEST_METRICS_NAMES[] = {"LocalTimeMs", "RemoteTimeMs", "RequestQueueTimeMs", "RequestQueueTimeMs",
      "RequestsPerSec", "ResponseQueueTimeMs", "ResponseSendTimeMs", "ThrottleTimeMs", "TotalTimeMs"};
  public static final String REQUEST_METRICS_REQUESTS[] =
      {"ApiVersions", "ControlledShutdown", "CreateTopics", "DeleteTopics", "DescribeGroups", "Fetch", "FetchConsumer", "FetchFollower",
          "GroupCoordinator", "Heartbeat", "JoinGroup", "LeaderAndIsr", "LeaveGroup", "ListGroups", "Metadata", "OffsetCommit",
          "OffsetFetch", "Offsets", "Produce", "SaslHandshake", "StopReplica", "SyncGroup", "UpdateMetadata"};

  public static final String NETWORK_PROCESSOR_AVG_IDLE_PERCENT = "kafka.network:type=SocketServer,name=NetworkProcessorAvgIdlePercent";

  // kafka.server
  public static final String BROKER_BYTES_IN_PER_SEC = "kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec";
  public static final String TOPIC_BYTES_IN_PER_SEC = "kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec,topic={}";
  public static final String BROKER_BYTES_OUT_PER_SEC = "kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec";
  public static final String TOPIC_BYTES_OUT_PER_SEC = "kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec,topic={}";
  public static final String BROKER_BYTES_REJECTERD_PER_SEC = "kafka.server:type=BrokerTopicMetrics,name=BytesRejectedPerSec";
  public static final String TOPIC_BYTES_REJECTERD_PER_SEC = "kafka.server:type=BrokerTopicMetrics,name=BytesRejectedPerSec,topic={}";
  public static final String BROKER_FAILED_FETCH_REQUESTS_PER_SEC = "kafka.server:type=BrokerTopicMetrics,name=FailedFetchRequestsPerSec";
  public static final String TOPIC_FAILED_FETCH_REQUESTS_PER_SEC =
      "kafka.server:type=BrokerTopicMetrics,name=FailedFetchRequestsPerSec,topic={}";
  public static final String BROKER_FAILED_PRODUCE_REQUESTS_PER_SEC =
      "kafka.server:type=BrokerTopicMetrics,name=FailedProduceRequestsPerSec";
  public static final String TOPIC_FAILED_PRODUCE_REQUESTS_PER_SEC =
      "kafka.server:type=BrokerTopicMetrics,name=FailedProduceRequestsPerSec,topic={}";
  public static final String BROKER_MESSAGES_IN_PER_SEC = "kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec";
  public static final String TOPIC_MESSAGES_IN_PER_SEC = "kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec,topic={}";
  public static final String TOTAL_FETCH_REQUESTS_PER_SEC = "kafka.server:type=BrokerTopicMetrics,name=TotalFetchRequestsPerSec";
  public static final String TOTAL_PRODUCE_REQUESTS_PER_SEC = "kafka.server:type=BrokerTopicMetrics,name=TotalProduceRequestsPerSec";

  public static final String CONSUMER_EXPIRES_PER_SEC = "kafka.server:type=DelayedFetchMetrics,name=ExpiresPerSec,fetcherType=consumer";
  public static final String FOLLOWER_EXPIRES_PER_SEC = "kafka.server:type=DelayedFetchMetrics,name=ExpiresPerSec,fetcherType=follower";

  public static final String DELAYED_OPERATIONS_FETCH_NUM =
      "kafka.server:type=DelayedOperationPurgatory,name=NumDelayedOperations,delayedOperation=Fetch";
  public static final String DELAYED_OPERATIONS_HEARTBEAT_NUM =
      "kafka.server:type=DelayedOperationPurgatory,name=NumDelayedOperations,delayedOperation=Heartbeat";
  public static final String DELAYED_OPERATIONS_PRODUCE_NUM =
      "kafka.server:type=DelayedOperationPurgatory,name=NumDelayedOperations,delayedOperation=Produce";
  public static final String DELAYED_OPERATIONS_REBALANCE_NUM =
      "kafka.server:type=DelayedOperationPurgatory,name=NumDelayedOperations,delayedOperation=Rebalance";
  public static final String DELAYED_OPERATIONS_TOPIC_NUM =
      "kafka.server:type=DelayedOperationPurgatory,name=NumDelayedOperations,delayedOperation=topic";

  public static final String DELAYED_OPERATIONS_FETCH_PURGATORY_SIZE =
      "kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=Fetch";
  public static final String DELAYED_OPERATIONS_HEARTBEAT_PURGATORY_SIZE =
      "kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=Heartbeat";
  public static final String DELAYED_OPERATIONS_PRODUCE_PURGATORY_SIZE =
      "kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=Produce";
  public static final String DELAYED_OPERATIONS_REBALANCE_PURGATORY_SIZE =
      "kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=Rebalance";
  public static final String DELAYED_OPERATIONS_TOPIC_PURGATORY_SIZE =
      "kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=topic";

  public static final String FETCH_DELAY_QUEUE_SIZE = "kafka.server:type=Fetch";
  public static final String PRODUCE_DELAY_QUEUE_SIZE = "kafka.server:type=Produce";

  public static final String CONSUMER_LAG = "kafka.server:type=FetcherLagMetrics,name=ConsumerLag,clientId={},topic={},partition={}";

  public static final String FETCHER_STATS_BYTES_PER_SEC =
      "kafka.server:type=FetcherStats,name=BytesPerSec,clientId={},brokerHost={},brokerPort={}";
  public static final String FETCHER_STATS_REQUESTS_PER_SEC =
      "kafka.server:type=FetcherStats,name=RequestsPerSec,clientId={},brokerHost={},brokerPort={}";

  public static final String REQUEST_HANDLER_AVG_IDLE_PERCENT =
      "kafka.server:type=KafkaRequestHandlerPool,name=RequestHandlerAvgIdlePercent";

  public static final String BROKER_STATE = "kafka.server:type=KafkaServer,name=BrokerState";
  public static final String CLUSTER_ID = "kafka.server:type=KafkaServer,name=ClusterId";

  public static final String LEADER_REPLICATION = "kafka.server:type=LeaderReplication";

  public static final String REPLICA_MAX_LAG = "kafka.server:type=ReplicaFetcherManager,name=MaxLag,clientId=Replica";
  public static final String REPLICA_MIN_FETCH_RATE = "kafka.server:type=ReplicaFetcherManager,name=MinFetchRate,clientId=Replica";
  public static final String REPLICA_ISR_EXPANDS_PER_SEC = "kafka.server:type=ReplicaManager,name=IsrExpandsPerSec";
  public static final String REPLICA_ISR_SHRINKS_PER_SEC = "kafka.server:type=ReplicaManager,name=IsrShrinksPerSec";
  public static final String REPLICA_LEADER_COUNT = "kafka.server:type=ReplicaManager,name=LeaderCount";
  public static final String REPLICA_PARTITION_COUNT = "kafka.server:type=ReplicaManager,name=PartitionCount";
  public static final String REPLICA_UNDERREPLICATED_PARTITIONS = "kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions";

  // kafka.server:type=SessionExpireListener
  public static final String ZK_AUTHFAILURES_PER_SEC = "kafka.server:type=SessionExpireListener,name=ZooKeeperAuthFailuresPerSec";
  public static final String ZK_DISCONNECTS_PER_SEC = "kafka.server:type=SessionExpireListener,name=ZooKeeperDisconnectsPerSec";
  public static final String ZK_EXPIRES_PER_SEC = "kafka.server:type=SessionExpireListener,name=ZooKeeperExpiresPerSec";
  public static final String ZK_RO_CONNECTS_PER_SEC = "kafka.server:type=SessionExpireListener,name=ZooKeeperReadOnlyConnectsPerSec";
  public static final String ZK_SASL_AUTHS_PER_SEC = "kafka.server:type=SessionExpireListener,name=ZooKeeperSaslAuthenticationsPerSec";
  public static final String ZK_SYNC_CONNECTIONS_PER_SEC = "kafka.server:type=SessionExpireListener,name=ZooKeeperSyncConnectsPerSec";

  // app-info
  public static final String APP_INFO = "kafka.server:type=app-info,id=0";

  public static final String REPLICA_FETCHER_METRICS = "kafka.server:type=replica-fetcher-metrics,broker-id={},fetcher-id={}";
  public static final String SOCKET_SERVER_METRICS = "kafka.server:type=socket-server-metrics,networkProcessor={}";

  public MeterMetric getItem(MBeanServerConnection mbsc, Optional<String> topicName, String metricName){
    return getBrokerTopicMetrics(mbsc, metricName, topicName);
  }
  public MeterMetric getBytesInPerSec(MBeanServerConnection mbsc, Optional<String> topicName) {
    return getBrokerTopicMetrics(mbsc, "BytesInPerSec", topicName);
  }

  public MeterMetric getBytesOutPerSec(MBeanServerConnection mbsc, Optional<String> topicName) {
    return getBrokerTopicMetrics(mbsc, "BytesOutPerSec", topicName);
  }

  public MeterMetric getBytesRejectedPerSec(MBeanServerConnection mbsc, Optional<String> topicName) {
    return getBrokerTopicMetrics(mbsc, "BytesRejectedPerSec", topicName);
  }

  public MeterMetric getFailedFetchRequestsPerSec(MBeanServerConnection mbsc, Optional<String> topicName) {
    return getBrokerTopicMetrics(mbsc, "FailedFetchRequestsPerSec", topicName);
  }

  public MeterMetric getFailedProduceRequestsPerSec(MBeanServerConnection mbsc, Optional<String> topicName) {
    return getBrokerTopicMetrics(mbsc, "FailedProduceRequestsPerSec", topicName);
  }

  public MeterMetric getMessagesInPerSec(MBeanServerConnection mbsc, Optional<String> topicName) {
    return getBrokerTopicMetrics(mbsc, "MessagesInPerSec", topicName);
  }

  private MeterMetric getBrokerTopicMetrics(MBeanServerConnection mbsc, String metricName, Optional<String> topicName) {
    return getMeterMetric(mbsc, getObjectName(metricName, topicName));
  }

  private MeterMetric getMeterMetric(MBeanServerConnection mbsc, ObjectName objectName) {
    String[] attributes = {"Count", "MeanRate", "OneMinuteRate", "FiveMinuteRate", "FifteenMinuteRate"};
    AttributeList attributeList = null;
    try {
      attributeList = mbsc.getAttributes(objectName, attributes);
    } catch (Exception e) {
      LOG.warn("getMeterMetric failed! " + e.getMessage());
      return new MeterMetric(0L, 0D, 0D, 0D, 0D);
    }
    return new MeterMetric(getLongValue(attributeList, attributes[0]), getDoubleValue(attributeList, attributes[1]),
        getDoubleValue(attributeList, attributes[2]), getDoubleValue(attributeList, attributes[3]),
        getDoubleValue(attributeList, attributes[4]));
  }

  private ObjectName getObjectName(String metricName, Optional<String> topicName) {
    ObjectName objectName = null;
    try {
      if (topicName.isPresent()) {
        objectName = new ObjectName("kafka.server:type=BrokerTopicMetrics,name=" + metricName + ",topic=" + topicName.get());
      } else {
        objectName = new ObjectName("kafka.server:type=BrokerTopicMetrics,name=" + metricName);
      }
    } catch (MalformedObjectNameException e) {
      LOG.error("Get ObjectName error! " + e.getMessage());
    }
    return objectName;
  }

  private Double getDoubleValue(AttributeList attributes, String name) {
    List<Attribute> _attributes = attributes.asList();
    for (Attribute attr : _attributes) {
      if (attr.getName().equalsIgnoreCase(name)) {
        return (Double) attr.getValue();
      }
    }
    return 0D;
  }

  private Long getLongValue(AttributeList attributes, String name) {
    List<Attribute> _attributes = attributes.asList();
    for (Attribute attr : _attributes) {
      if (attr.getName().equalsIgnoreCase(name)) {
        return (Long) attr.getValue();
      }
    }
    return 0L;
  }
} public class BrokerInfo {
	private int bid;
	private String host;
	private int port;
	private int jmxPort;
	private long timestamp;
	private int version;

	public BrokerInfo() {

	}

	public BrokerInfo(int bid, String host, int port) {
		super();
		this.bid = bid;
		this.host = host;
		this.port = port;
	}

	public int getBid() {
		return bid;
	}

	public void setBid(int bid) {
		this.bid = bid;
	}

	public String getHost() {
		return host;
	}

	public void setHost(String host) {
		this.host = host;
	}

	public int getPort() {
		return port;
	}

	public void setPort(int port) {
		this.port = port;
	}

	@Override
	public String toString() {
		return "BrokerInfo [bid=" + bid + ", host=" + host + ", port=" + port + "]";
	}

	public int getJmxPort() {
		return jmxPort;
	}

	public void setJmxPort(int jmx_port) {
		this.jmxPort = jmx_port;
	}

	public long getTimestamp() {
		return timestamp;
	}

	public void setTimestamp(long timestamp) {
		this.timestamp = timestamp;
	}

	public int getVersion() {
		return version;
	}

	public void setVersion(int version) {
		this.version = version;
	}

} public class KafkaInfo {
	private String GroupName;
	private List<BrokerInfo> brokers;
	private List<OffsetInfo> offsets;

	public KafkaInfo() {
		super();
	}

	public KafkaInfo(String name, List<BrokerInfo> brokers, List<OffsetInfo> offsets) {
		super();
		this.GroupName = name;
		this.brokers = brokers;
		this.offsets = offsets;
	}

	public String getName() {
		return GroupName;
	}

	public void setName(String name) {
		this.GroupName = name;
	}

	public List<BrokerInfo> getBrokers() {
		return brokers;
	}

	public void setBrokers(List<BrokerInfo> brokers) {
		this.brokers = brokers;
	}

	public List<OffsetInfo> getOffsets() {
		return offsets;
	}

	public void setOffsets(List<OffsetInfo> offsets) {
		this.offsets = offsets;
	}

	@Override
	public String toString() {
		return "KafkaInfo [GroupName=" + GroupName + ", brokers=" + brokers + ", offsets=" + offsets + "]";
	}

} public class CommonUtils {
	private static Logger LOG = LoggerFactory.getLogger(CommonUtils.class);

	public static void sortByTimestampThenPartition(List<OffsetPoints> offsetPointsList) {
		Collections.sort(offsetPointsList, new Comparator<OffsetPoints>() {

			@Override
			public int compare(OffsetPoints o1, OffsetPoints o2) {
				int flag = o1.getTimestamp().compareTo(o2.getTimestamp());
				if (flag == 0) {
					return o1.getPartition().compareTo(o2.getPartition());
				} else {
					return flag;
				}
			}
		});
	}

	public static String loadFileContent(String filePath) {
		String contents = null;
		try {
			while (!(new File(filePath)).canRead()) {
				// TODO
				Thread.sleep(10);
			}
			contents = new String(Files.readAllBytes(Paths.get(filePath)));
		} catch (Exception e) {
			LOG.error("load file" + filePath + " failed!", e);
		}
		if (contents == null) {
			throw new RuntimeException("file is empty! Pls check file: " + filePath);
		}
		return contents;
	}

	public static URI getURI(String urlStr) {
		URL url = null;
		URI uri = null;
		try {
			url = new URL(urlStr);
			uri = new URI(url.getProtocol(), url.getHost(), url.getPath(), url.getQuery(), null);
		} catch (Exception e) {
			LOG.error("URL error!");
			// TODO
			return null;
		}
		return uri;
	}
} public class ZKUtils {
	private static Logger LOG = LoggerFactory.getLogger(ZKUtils.class);

	private static ZkClient zkClient = null;
	private static ZkConnection zkConnection = null;
	private static ZkUtils zkUtilsFromKafka = null;

	public static void init(String zkHosts, int zkSessionTimeout, int zkConnectionTimeout) {
		try {
			LOG.debug("init ZKUtil with " + zkHosts + " - " + zkSessionTimeout + " - " + zkConnectionTimeout);
			if (zkConnection == null) {
				zkConnection = new ZkConnection(zkHosts);
			}
			if (zkClient == null) {
				zkClient = new ZkClient(zkConnection, zkConnectionTimeout, new ZkSerializer() {

					@Override
					public byte[] serialize(Object paramObject) throws ZkMarshallingError {
						return ZKStringSerializer.serialize(paramObject);
					}

					@Override
					public Object deserialize(byte[] paramArrayOfByte) throws ZkMarshallingError {
						return ZKStringSerializer.deserialize(paramArrayOfByte);
					}

				});
			}
			if (zkUtilsFromKafka == null) {
				zkUtilsFromKafka = new ZkUtils(zkClient, zkConnection, false);
			}
		} catch (Exception e) {
			throw new RuntimeException("Init ZKUtils failed! " + e.getMessage());
		}
	}

	public static ZkClient getZKClient() {
		return zkClient;
	}

	public static ZkUtils getZKUtilsFromKafka() {
		return zkUtilsFromKafka;
	}

	public static Set<TopicAndPartition> getAllPartitions() {
		return JavaConversions.setAsJavaSet(zkUtilsFromKafka.getAllPartitions());
	}

	public static Map<String, List<Integer>> getPartitionsForTopics(List<String> topics) {
		Map<String, List<Integer>> topicPartitions = new HashMap<String, List<Integer>>();
		Map<String, Map<Integer, List<Integer>>> ps4Topics = getPartitionAssignmentForTopics(topics);
		ps4Topics.forEach((topic, PartitionReplica) -> {
			List<Integer> partitions = new ArrayList<Integer>();
			PartitionReplica.forEach((partition, replicas) -> {
				partitions.add(partition);
			});
			Collections.sort(partitions, new Comparator<Integer>() {
				@Override
				public int compare(Integer pid1, Integer pid2) {
					return pid1 < pid2 ? 1 : -1;
				}
			});
			topicPartitions.put(topic, partitions);
		});
		return topicPartitions;

	}

	public static Map<String, Map<Integer, List<Integer>>> getPartitionAssignmentForTopics(List<String> topics) {
		Map<String, Map<Integer, List<Integer>>> ret = new HashMap<String, Map<Integer, List<Integer>>>();
		topics.forEach(topic -> {
			String data = readDataMaybeNull(ZkUtils.BrokerTopicsPath() + "/" + topic).getData();
			if (data == null) {
				ret.put(topic, null);
			} else {
				JSONObject json = new JSONObject(data).getJSONObject("partitions");
				Map<Integer, List<Integer>> replicaMap = new HashMap<Integer, List<Integer>>();
				for (String pid : json.keySet()) {
					JSONArray bidsJson = json.getJSONArray(pid);
					List<Integer> replicas = new ArrayList<Integer>();
					bidsJson.forEach(bid -> {
						replicas.add((Integer) bid);
					});
					replicaMap.put(Integer.parseInt(pid), replicas);
				}
				ret.put(topic, replicaMap);
			}
		});
		return ret;
	}

	public static List<String> getAllTopics() {
		return JavaConversions.seqAsJavaList(zkUtilsFromKafka.getAllTopics());
	}

	public static Option<Broker> getBrokerInfo(int brokerId) {
		// Option[Broker] = zkUtilsFromKafka.getBrokerInfo(zkClient, brokerId)
		return zkUtilsFromKafka.getBrokerInfo(brokerId);
	}

	public static List<String> getConsumersInGroup(String group) {
		return JavaConversions.seqAsJavaList(zkUtilsFromKafka.getConsumersInGroup(group));

	}

	public static List<String> parseTopicsData(String jsonData) {
		return JavaConversions.seqAsJavaList(ZkUtils.parseTopicsData(jsonData));
	}

	public static boolean pathExists(String path) {
		return zkUtilsFromKafka.pathExists(path);
	}

	public static List<String> getChildren(String path) {
		List<String> children = null;
		try {
			children = zkClient.getChildren(path);
		} catch (Exception e) { // that should be
								// {org.apache.zookeeper.KeeperException$NoNodeException}
			children = new ArrayList<String>();
		}
		return children;
	}

	public static ZkDataAndStat readDataMaybeNull(String path) {
		Stat stat = new Stat();
		String data = null;
		try {
			data = zkClient.readData(path, stat);
		} catch (Exception e) {
			LOG.warn("Path: " + path + " do not exits in ZK!" + e.getMessage());
		}
		return new ZkDataAndStat(data, stat);
	}

	public static List<BrokerInfo> getBrokers() throws Exception {
		List<BrokerInfo> kafkaHosts = new ArrayList<BrokerInfo>();
		List<String> ids = getChildren("/brokers/ids");
		for (String id : ids) {
			try {
				String brokerInfo = new String(readDataMaybeNull("/brokers/ids/" + id).getData());

				BrokerInfo bi = new BrokerInfo();
				bi.setBid(Integer.parseInt(id));
				JSONObject jsonObj = new JSONObject(brokerInfo);
				if (jsonObj.has("host")) {
					bi.setHost(jsonObj.get("host").toString());
				}
				if (jsonObj.has("port")) {
					bi.setPort(Integer.parseInt(jsonObj.get("port").toString()));
				}
				if (jsonObj.has("jmx_port")) {
					bi.setJmxPort(Integer.parseInt(jsonObj.get("jmx_port").toString()));
				}
				if (jsonObj.has("version")) {
					bi.setVersion(Integer.parseInt(jsonObj.get("version").toString()));
				}
				if (jsonObj.has("timestamp")) {
					bi.setTimestamp(Long.parseLong(jsonObj.get("timestamp").toString()));
				}
				kafkaHosts.add(bi);
			} catch (Exception e) {
				LOG.error("Zookeeper borker getting exception {}", e);
			}
		}
		return kafkaHosts;
	}

	// TODO for those invalid consumer id content, is that a concerned part?
	// public static Map<String, List<ConsumerThreadId>>
	// getConsumersPerTopic(String
	// group,
	// boolean excludeInternalTopics) {
	// List<String> consumers = getChildren(ZkUtils.ConsumersPath() + "/" +
	// group +
	// "/ids");
	// Map<String, List<ConsumerThreadId>> consumersPerTopicMap = new
	// HashMap<String, List<ConsumerThreadId>>();
	// for (String consumer : consumers) {
	// TopicCount topicCount = constructTopicCount(group, consumer,
	// zkUtilsFromKafka, excludeInternalTopics);
	// }
	//
	// return null;
	//
	// }
	//
	// private static TopicCount constructTopicCount(String group, String
	// consumerId, ZkUtils zkUtilsFromKafka,
	// boolean excludeInternalTopics) {
	// ZkDataAndStat consumerData = readDataMaybeNull(
	// ZkUtils.ConsumersPath() + "/" + group + "/ids" + "/" + consumerId);
	// String topicCountString = consumerData.getData();
	// String subscriptionPattern = null;
	// Map<String, Integer> topMap = null;
	// //
	// {"version":1,"subscription":{"EC2_SHOPPINGCART":1},"pattern":"static","timestamp":"1498610861408"}
	// if (topicCountString != null) {
	// JSONObject topicCountJson = new JSONObject(topicCountString);
	//
	// }
	// return null;
	// }
	public static void close() {
		try {
			if (zkUtilsFromKafka != null) {
				zkUtilsFromKafka.close();
				zkUtilsFromKafka = null;
			}
			if (zkClient != null) {
				zkClient.close();
				zkClient = null;
			}
			if (zkConnection != null) {
				zkConnection.close();
				zkConnection = null;
			}
		} catch (InterruptedException e) {
			LOG.error("ZKUtils close() error! ", e);
		}
	}
} public interface Ielasticsearch {

  void bulkIndex(JSONObject data, String docType, String indexPrefix);

  List<OffsetPoints> offsetHistory(String indexPrefix, String docType, String group, String topic);

  List<OffsetPoints> scrollsSearcher(OffsetHistoryQueryParams params, String docType, String indexPrefix);

  boolean check();

  void close();

}
import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.management.MBeanServerConnection;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.chickling.kmanager.alert.TaskContent;
import com.chickling.kmanager.alert.TaskHandler;
import com.chickling.kmanager.alert.TaskManager;
import com.chickling.kmanager.alert.WorkerThreadFactory;
import com.chickling.kmanager.config.AppConfig;
import com.chickling.kmanager.core.OffsetGetter;
import com.chickling.kmanager.core.ZKOffsetGetter;
import com.chickling.kmanager.core.db.ElasticsearchOffsetDB;
import com.chickling.kmanager.core.db.OffsetDB;
import com.chickling.kmanager.email.EmailSender;
import com.chickling.kmanager.jmx.FormatedMeterMetric;
import com.chickling.kmanager.jmx.JMXExecutor;
import com.chickling.kmanager.jmx.KafkaJMX;
import com.chickling.kmanager.jmx.KafkaMetrics;
import com.chickling.kmanager.model.BrokerInfo;
import com.chickling.kmanager.model.KafkaInfo;
import com.chickling.kmanager.utils.CommonUtils;
import com.chickling.kmanager.utils.ZKUtils;
import com.chickling.kmanager.utils.elasticsearch.Ielasticsearch;
import com.google.gson.Gson;
package com.chickling.kmanager.initialize;





/**
 * @author Hulva Luva.H
 * @since 2017-07-13
 *
 */
public class SystemManager {
  private static Logger LOG = LoggerFactory.getLogger(SystemManager.class);

  // TODO schedule pool size? any other schedule?
  private static ScheduledExecutorService scheduler = null;

  private static ExecutorService worker;

  public static final int DEFAULT_THREAD_POOL_SIZE = Runtime.getRuntime().availableProcessors();

  private static final ExecutorService kafkaInfoCollectAndSavePool = Executors
      .newFixedThreadPool(DEFAULT_THREAD_POOL_SIZE, new WorkerThreadFactory("KafkaInfo Collector"));

  public static final String JMX_METRIC_ES_DOC_TYPE = "jmxMetrics";

  public static BlockingQueue<KafkaInfo> offsetInfoCacheQueue;

  public static OffsetDB<Ielasticsearch> db = null;