// Path: src/main/java/org/smartloli/kafka/eagle/common/domain/HostsDomain.java
// public class HostsDomain {
// 
// 	private String host;
// 	private int port;
// 
// 	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 new Gson().toJson(this);
// 	}
// 
// }
// 
// Path: src/main/java/org/smartloli/kafka/eagle/common/domain/KafkaSqlDomain.java
// public class KafkaSqlDomain {
// 
// 	private List<Integer> partition = new ArrayList<>();
// 	private String sql;
// 	private String metaSql;
// 	private JSONObject schema = new JSONObject();
// 	private String tableName;
// 	private String topic;
// 	private boolean status;
// 	private List<HostsDomain> seeds = new ArrayList<>();
// 
// 	public String getTopic() {
// 		return topic;
// 	}
// 
// 	public void setTopic(String topic) {
// 		this.topic = topic;
// 	}
// 
// 	public String getMetaSql() {
// 		return metaSql;
// 	}
// 
// 	public void setMetaSql(String metaSql) {
// 		this.metaSql = metaSql;
// 	}
// 
// 	public List<HostsDomain> getSeeds() {
// 		return seeds;
// 	}
// 
// 	public void setSeeds(List<HostsDomain> seeds) {
// 		this.seeds = seeds;
// 	}
// 
// 	public JSONObject getSchema() {
// 		return schema;
// 	}
// 
// 	public void setSchema(JSONObject schema) {
// 		this.schema = schema;
// 	}
// 
// 	public String getTableName() {
// 		return tableName;
// 	}
// 
// 	public void setTableName(String tableName) {
// 		this.tableName = tableName;
// 	}
// 
// 	public List<Integer> getPartition() {
// 		return partition;
// 	}
// 
// 	public void setPartition(List<Integer> partition) {
// 		this.partition = partition;
// 	}
// 
// 	public String getSql() {
// 		return sql;
// 	}
// 
// 	public void setSql(String sql) {
// 		this.sql = sql;
// 	}
// 
// 	public boolean isStatus() {
// 		return status;
// 	}
// 
// 	public void setStatus(boolean status) {
// 		this.status = status;
// 	}
// 
// 	@Override
// 	public String toString() {
// 		return new Gson().toJson(this);
// 	}
// 
// }


import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.smartloli.kafka.eagle.common.domain.HostsDomain;
import org.smartloli.kafka.eagle.common.domain.KafkaSqlDomain;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import kafka.api.FetchRequest;
import kafka.api.FetchRequestBuilder;
import kafka.api.PartitionOffsetRequestInfo;
import kafka.common.ErrorMapping;
import kafka.common.TopicAndPartition;
import kafka.javaapi.FetchResponse;
import kafka.javaapi.OffsetResponse;
import kafka.javaapi.PartitionMetadata;
import kafka.javaapi.TopicMetadata;
import kafka.javaapi.TopicMetadataRequest;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.message.MessageAndOffset;
/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.smartloli.kafka.eagle.core.sql.execute;





/**
 * Kafka underlying consumer API interface.
 *
 * @author smartloli.
 *
 *         Created by Mar 14, 2016
 */
public class SimpleKafkaConsumer {
	private static Logger LOG = LoggerFactory.getLogger(SimpleKafkaConsumer.class);
	private List<HostsDomain> m_replicaBrokers = new ArrayList<HostsDomain>();
	private static int buff_size = 64 * 1024;
	private static int fetch_size = 1000 * 1000 * 1000;
	private static int timeout = 100000;
	private static int maxSize = 5000;

	public SimpleKafkaConsumer() {
		m_replicaBrokers = new ArrayList<HostsDomain>();
	}

	public static List<JSONArray> start(KafkaSqlDomain kafkaSql) {
		List<JSONArray> messages = new ArrayList<>();
		List<HostsDomain> seeds = kafkaSql.getSeeds();
		for (int partition : kafkaSql.getPartition()) {
			messages.add(consumer(partition, kafkaSql.getTopic(), seeds));
		}
		return messages;
	}

	public static JSONArray consumer(int _partition, String _topic, List<HostsDomain> seeds) {
		JSONArray msg = null;
		SimpleKafkaConsumer example = new SimpleKafkaConsumer();
		// Max read number
		long maxReads = 3L;
		// To subscribe to the topic
		String topic = _topic;
		// Find partition
		int partition = _partition;
		try {
			msg = example.run(maxReads, topic, partition, seeds);
// Path: src/main/java/org/smartloli/kafka/eagle/common/domain/HostsDomain.java
// public class HostsDomain {
// 
// 	private String host;
// 	private int port;
// 
// 	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 new Gson().toJson(this);
// 	}
// 
// }
// 
// Path: src/main/java/org/smartloli/kafka/eagle/common/domain/KafkaSqlDomain.java
// public class KafkaSqlDomain {
// 
// 	private List<Integer> partition = new ArrayList<>();
// 	private String sql;
// 	private String metaSql;
// 	private JSONObject schema = new JSONObject();
// 	private String tableName;
// 	private String topic;
// 	private boolean status;
// 	private List<HostsDomain> seeds = new ArrayList<>();
// 
// 	public String getTopic() {
// 		return topic;
// 	}
// 
// 	public void setTopic(String topic) {
// 		this.topic = topic;
// 	}
// 
// 	public String getMetaSql() {
// 		return metaSql;
// 	}
// 
// 	public void setMetaSql(String metaSql) {
// 		this.metaSql = metaSql;
// 	}
// 
// 	public List<HostsDomain> getSeeds() {
// 		return seeds;
// 	}
// 
// 	public void setSeeds(List<HostsDomain> seeds) {
// 		this.seeds = seeds;
// 	}
// 
// 	public JSONObject getSchema() {
// 		return schema;
// 	}
// 
// 	public void setSchema(JSONObject schema) {
// 		this.schema = schema;
// 	}
// 
// 	public String getTableName() {
// 		return tableName;
// 	}
// 
// 	public void setTableName(String tableName) {
// 		this.tableName = tableName;
// 	}
// 
// 	public List<Integer> getPartition() {
// 		return partition;
// 	}
// 
// 	public void setPartition(List<Integer> partition) {
// 		this.partition = partition;
// 	}
// 
// 	public String getSql() {
// 		return sql;
// 	}
// 
// 	public void setSql(String sql) {
// 		this.sql = sql;
// 	}
// 
// 	public boolean isStatus() {
// 		return status;
// 	}
// 
// 	public void setStatus(boolean status) {
// 		this.status = status;
// 	}
// 
// 	@Override
// 	public String toString() {
// 		return new Gson().toJson(this);
// 	}
// 
// }


// Path: src/main/java/org/smartloli/kafka/eagle/core/sql/execute/SimpleKafkaConsumer.java
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.smartloli.kafka.eagle.common.domain.HostsDomain;
import org.smartloli.kafka.eagle.common.domain.KafkaSqlDomain;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import kafka.api.FetchRequest;
import kafka.api.FetchRequestBuilder;
import kafka.api.PartitionOffsetRequestInfo;
import kafka.common.ErrorMapping;
import kafka.common.TopicAndPartition;
import kafka.javaapi.FetchResponse;
import kafka.javaapi.OffsetResponse;
import kafka.javaapi.PartitionMetadata;
import kafka.javaapi.TopicMetadata;
import kafka.javaapi.TopicMetadataRequest;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.message.MessageAndOffset;

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.smartloli.kafka.eagle.core.sql.execute;





/**
 * Kafka underlying consumer API interface.
 *
 * @author smartloli.
 *
 *         Created by Mar 14, 2016
 */
public class SimpleKafkaConsumer {
	private static Logger LOG = LoggerFactory.getLogger(SimpleKafkaConsumer.class);
	private List<HostsDomain> m_replicaBrokers = new ArrayList<HostsDomain>();
	private static int buff_size = 64 * 1024;
	private static int fetch_size = 1000 * 1000 * 1000;
	private static int timeout = 100000;
	private static int maxSize = 5000;

	public SimpleKafkaConsumer() {
		m_replicaBrokers = new ArrayList<HostsDomain>();
	}

	public static List<JSONArray> start(KafkaSqlDomain kafkaSql) {
		List<JSONArray> messages = new ArrayList<>();
		List<HostsDomain> seeds = kafkaSql.getSeeds();
		for (int partition : kafkaSql.getPartition()) {
			messages.add(consumer(partition, kafkaSql.getTopic(), seeds));
		}
		return messages;
	}

	public static JSONArray consumer(int _partition, String _topic, List<HostsDomain> seeds) {
		JSONArray msg = null;
		SimpleKafkaConsumer example = new SimpleKafkaConsumer();
		// Max read number
		long maxReads = 3L;
		// To subscribe to the topic
		String topic = _topic;
		// Find partition
		int partition = _partition;
		try {
			msg = example.run(maxReads, topic, partition, seeds);
		} catch (Exception e) {