ksql细节改动

kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java

@@ -33,8 +33,6 @@
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 
 import javax.management.MBeanServerConnection;
 import javax.management.ObjectName;
@@ -67,7 +65,6 @@
 import org.slf4j.LoggerFactory;
 import org.smartloli.kafka.eagle.common.constant.JmxConstants.BrokerServer;
 import org.smartloli.kafka.eagle.common.constant.JmxConstants.KafkaServer8;
-import org.smartloli.kafka.eagle.common.constant.OdpsSqlParser;
 import org.smartloli.kafka.eagle.common.protocol.BrokersInfo;
 import org.smartloli.kafka.eagle.common.protocol.DisplayInfo;
 import org.smartloli.kafka.eagle.common.protocol.HostsInfo;
@@ -90,6 +87,7 @@
 import com.alibaba.fastjson.JSONObject;
 
 import kafka.zk.KafkaZkClient;
+import org.smartloli.kafka.eagle.core.sql.tool.OdpsSqlParser;
 import scala.Option;
 import scala.Tuple2;
 import scala.collection.JavaConversions;
@@ -339,16 +337,11 @@ public Map<String, List<String>> getConsumers(String clusterAlias, DisplayInfo p
 
 	/**
 	 * According to group, topic and partition to get offset from zookeeper.
-	 * 
-	 * @param topic
-	 *            Filter topic.
-	 * @param group
-	 *            Filter group.
-	 * @param partition
-	 *            Filter partition.
+	 *
+	 * @param topic     Filter topic.
+	 * @param group     Filter group.
+	 * @param partition Filter partition.
 	 * @return OffsetZkInfo.
-	 * 
-	 * @see org.smartloli.kafka.eagle.domain.OffsetZkInfo
 	 */
 	public OffsetZkInfo getOffset(String clusterAlias, String topic, String group, int partition) {
 		KafkaZkClient zkc = kafkaZKPool.getZkClient(clusterAlias);

kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/BrokerServiceImpl.java

@@ -17,16 +17,8 @@
  */
 package org.smartloli.kafka.eagle.core.factory.v2;
 
-import java.util.ArrayList;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Properties;
-import java.util.Set;
-import java.util.TreeSet;
 
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.admin.AdminClient;
@@ -81,10 +73,20 @@ public long topicNumbers(String clusterAlias) {
 		return topicList(clusterAlias).size();
 	}
 
-	/** Exclude kafka topic(__consumer_offsets). */
-	private void excludeTopic(List<String> topics) {
-		if (topics.contains(Kafka.CONSUMER_OFFSET_TOPIC)) {
-			topics.remove(Kafka.CONSUMER_OFFSET_TOPIC);
+	/**
+	 * Exclude kafka topic(__consumer_offsets).
+	 */
+	private void excludeTopic(List<String> topics, String clusterAlias, KafkaZkClient zkc) {
+
+		topics.remove(Kafka.CONSUMER_OFFSET_TOPIC);
+
+		Iterator<String> iterator = topics.iterator();
+		while (iterator.hasNext()) {
+			String topic = iterator.next();
+			if (!zkc.pathExists(BROKER_TOPICS_PATH + "/" + topic + "/partitions")) {
+				iterator.remove();
+				LOG.info("remove topic" + topic);
+			}
 		}
 	}
 
@@ -346,7 +348,7 @@ public List<String> topicList(String clusterAlias) {
 				if (zkc.pathExists(BROKER_TOPICS_PATH)) {
 					Seq<String> subBrokerTopicsPaths = zkc.getChildren(BROKER_TOPICS_PATH);
 					topics = JavaConversions.seqAsJavaList(subBrokerTopicsPaths);
-					excludeTopic(topics);
+					excludeTopic(topics, clusterAlias,zkc);
 				}
 			} catch (Exception e) {
 				LOG.error("Get topic list has error, msg is " + e.getCause().getMessage());

kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/execute/KafkaSqlParser.java

@@ -21,18 +21,13 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ConcurrentMap;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.smartloli.kafka.eagle.common.protocol.KafkaSqlInfo;
 import org.smartloli.kafka.eagle.core.factory.KafkaFactory;
 import org.smartloli.kafka.eagle.core.factory.KafkaService;
-import org.smartloli.kafka.eagle.core.factory.v2.BrokerFactory;
-import org.smartloli.kafka.eagle.core.factory.v2.BrokerService;
 import org.smartloli.kafka.eagle.core.sql.tool.JSqlUtils;
-
-import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.google.gson.Gson;
 
@@ -47,7 +42,6 @@ public class KafkaSqlParser {
 
     private final static Logger LOG = LoggerFactory.getLogger(KafkaSqlParser.class);
     private static KafkaService kafkaService = new KafkaFactory().create();
-    private static BrokerService brokerService = new BrokerFactory().create();
 
     public static String execute(String clusterAlias, String sql) {
         JSONObject status = new JSONObject();

kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/function/JSONFunction.java

@@ -20,6 +20,8 @@
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 
+import java.util.Objects;
+
 /**
  * Parse a JSONObject or a JSONArray in a kafka topic message using a custom
  * function.
@@ -32,19 +34,31 @@ public class JSONFunction {
 
 	/** Parse a JSONObject. */
 	public String JSON(String jsonObject, String key) {
-		JSONObject object = com.alibaba.fastjson.JSON.parseObject(jsonObject);
+		JSONObject object = null;
+		try {
+			Objects.requireNonNull(jsonObject);
+			object = com.alibaba.fastjson.JSON.parseObject(jsonObject);
+		} catch (Exception e) {
+			return null;
+		}
 		return object.getString(key);
 	}
 
 	/** Parse a JSONArray. */
 	public String JSONS(String jsonArray, String key) {
-		JSONArray object = com.alibaba.fastjson.JSON.parseArray(jsonArray);
-		JSONArray target = new JSONArray();
-		for (Object tmp : object) {
-			JSONObject result = (JSONObject) tmp;
-			JSONObject value = new JSONObject();
-			value.put(key, result.getString(key));
-			target.add(value);
+		JSONArray target = null;
+		try {
+			Objects.requireNonNull(jsonArray);
+			JSONArray object = com.alibaba.fastjson.JSON.parseArray(jsonArray);
+			target = new JSONArray();
+			for (Object tmp : object) {
+				JSONObject result = (JSONObject) tmp;
+				JSONObject value = new JSONObject();
+				value.put(key, result.getString(key));
+				target.add(value);
+			}
+		} catch (Exception e) {
+			return null;
 		}
 		return target.toJSONString();
 	}

kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/tool/JSqlUtils.java

@@ -17,21 +17,16 @@
  */
 package org.smartloli.kafka.eagle.core.sql.tool;
 
-import java.io.File;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.ResultSet;
-import java.sql.ResultSetMetaData;
-import java.sql.Statement;
+import java.sql.*;
 import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.smartloli.kafka.eagle.common.constant.JConstants;
 import org.smartloli.kafka.eagle.core.sql.common.JSqlMapData;
 
@@ -48,37 +43,56 @@
  */
 public class JSqlUtils {
 
+    private final static Logger LOG = LoggerFactory.getLogger(JSqlUtils.class);
+
     /**
-     * @param tabSchema : Table column,such as {"id":"integer","name":"varchar"}
-     * @param dataSets  : DataSets ,such as
-     *                  [{"id":1,"name":"aaa"},{"id":2,"name":"bbb"},{}...]
-     * @param sql       : such as "SELECT * FROM TBL"
-     * @return String
-     * @throws Exception : Throws an exception
+     * @param tabSchema
+     * @param dataSets
+     * @param sql
+     * @return
+     * @throws Exception
      */
     public static String query(JSONObject tabSchema, Map<String, List<List<String>>> dataSets, String sql) throws Exception {
-        String model = createTempJson();
+        Connection connection = null;
+        ResultSet result = null;
+        List<Map<String, Object>> ret = null;
+        Statement st = null;
+        try {
+            String model = createTempJson();
 
-        JSqlMapData.loadSchema(tabSchema, dataSets);
+            JSqlMapData.loadSchema(tabSchema, dataSets);
 
-        Class.forName(JConstants.KAFKA_DRIVER);
-        Properties info = new Properties();
-        info.setProperty("lex", "JAVA");
+            Class.forName(JConstants.KAFKA_DRIVER);
+            Properties info = new Properties();
+            info.setProperty("lex", "JAVA");
 
-        Connection connection = DriverManager.getConnection("jdbc:calcite:model=inline:" + model, info);
-        Statement st = connection.createStatement();
-        ResultSet result = st.executeQuery(sql);
-        ResultSetMetaData rsmd = result.getMetaData();
-        List<Map<String, Object>> ret = new ArrayList<Map<String, Object>>();
-        while (result.next()) {
-            Map<String, Object> map = new HashMap<String, Object>();
-            for (int i = 1; i <= rsmd.getColumnCount(); i++) {
-                map.put(rsmd.getColumnName(i), result.getString(rsmd.getColumnName(i)));
+            connection = DriverManager.getConnection("jdbc:calcite:model=inline:" + model, info);
+            st = connection.createStatement();
+            result = st.executeQuery(sql);
+            ResultSetMetaData rsmd = result.getMetaData();
+            ret = new ArrayList<Map<String, Object>>();
+            while (result.next()) {
+                Map<String, Object> map = new HashMap<String, Object>();
+                for (int i = 1; i <= rsmd.getColumnCount(); i++) {
+                    map.put(rsmd.getColumnName(i), result.getString(rsmd.getColumnName(i)));
+                }
+                ret.add(map);
+            }
+        } catch (ClassNotFoundException e) {
+            LOG.error(ExceptionUtils.getStackTrace(e));
+        } catch (SQLException e) {
+            LOG.error(ExceptionUtils.getStackTrace(e));
+        } finally {
+            if (st != null) {
+                st.close();
+            }
+            if (result != null) {
+                result.close();
+            }
+            if (connection != null) {
+                connection.close();
             }
-            ret.add(map);
         }
-        result.close();
-        connection.close();
         return new Gson().toJson(ret);
     }
 
@@ -100,7 +114,7 @@ public static String toJSONObject(List<JSONArray> dataSets) {
         return new Gson().toJson(results);
     }
 
-    private static String createTempJson() throws IOException {
+    private static String createTempJson() {
         JSONObject object = new JSONObject();
         object.put("version", "1.21.0");
         object.put("defaultSchema", "db");