Skip to content

ksql支持使用= #591

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aurorazl wants to merge 2 commits into
base: master
Choose a base branch
from
Open

ksql支持使用= #591

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f606f65
ksql支持使用=
Feb 27, 2022
cc6d190
consumers栏展示具体的topic名字，并支持搜索group和topic
Feb 27, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion efak-common/src/main/java/org/smartloli/kafka/eagle/common/constant/KSqlParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,16 @@
package org.smartloli.kafka.eagle.common.constant;

import org.apache.calcite.config.Lex;
import org.apache.calcite.rel.core.Collect;
import org.apache.calcite.sql.*;
import org.apache.calcite.sql.parser.SqlParser;
import org.smartloli.kafka.eagle.common.protocol.topic.TopicPartitionSchema;
import org.smartloli.kafka.eagle.common.util.LoggerUtils;
import org.smartloli.kafka.eagle.common.util.StrUtils;

import java.util.Arrays;
import java.util.Collections;

/**
* The client requests the t operation to parse the SQL and obtain the fields
* and conditions.
Expand Down Expand Up @@ -54,7 +58,7 @@ public static TopicPartitionSchema parserTopic(String sql) {
}

private static void parseNode(SqlNode sqlNode, TopicPartitionSchema tps) {
SqlKind sqlKind = sqlNode.getKind();
SqlKind sqlKind = sqlNode.getKind();
switch (sqlKind) {
case SELECT:
String topic = "";
Expand All @@ -76,6 +80,14 @@ private static void parseNode(SqlNode sqlNode, TopicPartitionSchema tps) {
tps.setTopic(topic);
tps.setPartitions(StrUtils.stringsConvertIntegers(partitions));
}
} else if (sqlWhere.getKind() == SqlKind.EQUALS) {
SqlBasicCall sqlBasicCall = (SqlBasicCall) sqlWhere;
if (sqlBasicCall.operands.length == 2) {
String[] partitions = Collections.singletonList(sqlBasicCall.operands[1].toString()).toArray(new String[0]);
tps.getTopicSchema().put(topic, StrUtils.stringsConvertIntegers(partitions));
tps.setTopic(topic);
tps.setPartitions(StrUtils.stringsConvertIntegers(partitions));
}
} else if (sqlWhere.getKind() == SqlKind.AND) {// two and
SqlBasicCall sqlBasicCall = (SqlBasicCall) sqlWhere;
if (sqlBasicCall.operands.length > 0) {
Expand Down
6 changes: 3 additions & 3 deletions efak-common/src/main/java/org/smartloli/kafka/eagle/common/protocol/ConsumerInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ public class ConsumerInfo extends BaseProtocol {

private int id;
private String group;
private int topics;
private String topics;
private String node;
private int activeTopics;
private int activeThreads;
Expand Down Expand Up @@ -65,11 +65,11 @@ public void setGroup(String group) {
this.group = group;
}

public int getTopics() {
public String getTopics() {
return topics;
}

public void setTopics(int topics) {
public void setTopics(String topics) {
this.topics = topics;
}

Expand Down
10 changes: 5 additions & 5 deletions ...src/main/java/org/smartloli/kafka/eagle/common/protocol/consumer/ConsumerSummaryInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@
public class ConsumerSummaryInfo extends BaseProtocol {
private String cluster;
private String group;
private int topicNumbers;
private String topics;
private String coordinator;
private int activeTopic;
private int activeThread;
Expand Down Expand Up @@ -74,12 +74,12 @@ public void setGroup(String group) {
this.group = group;
}

public int getTopicNumbers() {
return topicNumbers;
public String getTopics() {
return topics;
}

public void setTopicNumbers(int topicNumbers) {
this.topicNumbers = topicNumbers;
public void setTopics(String topics) {
this.topics = topics;
}

}
2 changes: 1 addition & 1 deletion efak-plugin/src/main/java/org/smartloli/kafka/eagle/plugin/util/JConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,7 +120,7 @@ public interface JConstants {

static String CREATE_TABLE_KE_CONSUMER_GROUP = "CREATE TABLE IF NOT EXISTS `ke_consumer_group` (`cluster` varchar(64) NOT NULL,`group` varchar(128) NOT NULL,`topic` varchar(128) NOT NULL,`status` int(11) DEFAULT NULL, PRIMARY KEY (`cluster`,`group`,`topic`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;";

static String CREATE_TABLE_KE_CONSUMER_GROUP_SUMMARY = "CREATE TABLE IF NOT EXISTS `ke_consumer_group_summary` (`cluster` varchar(64) NOT NULL,`group` varchar(128) NOT NULL,`topic_number` varchar(128) NOT NULL,`coordinator`varchar(128) DEFAULT NULL,`active_topic` int(11) DEFAULT NULL,`active_thread_total` int(11) DEFAULT NULL, PRIMARY KEY (`cluster`,`group`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;";
static String CREATE_TABLE_KE_CONSUMER_GROUP_SUMMARY = "CREATE TABLE IF NOT EXISTS `ke_consumer_group_summary` (`cluster` varchar(64) NOT NULL,`group` varchar(128) NOT NULL,`topics` varchar(256) NOT NULL,`coordinator`varchar(128) DEFAULT NULL,`active_topic` int(11) DEFAULT NULL,`active_thread_total` int(11) DEFAULT NULL, PRIMARY KEY (`cluster`,`group`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;";
static String CREATE_TABLE_KE_TOPIC_CONSUMER_GROUP_SUMMARY_VIEW = "CREATE OR REPLACE VIEW ke_topic_consumer_group_summary_view as select `cluster`, `topic`, count(distinct `group`) group_number , count(distinct `group`) active_group from `ke_consumer_group` where `status` = 0 group by `cluster`, `topic`";

static String CREATE_TABLE_KE_USER_ROLE = "CREATE TABLE IF NOT EXISTS `ke_user_role` (`id` bigint(20) NOT NULL AUTO_INCREMENT,`user_id` int(11) NOT NULL,`role_id` tinyint(4) NOT NULL,PRIMARY KEY (`id`)) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4";
Expand Down
4 changes: 2 additions & 2 deletions efak-web/src/main/java/org/smartloli/kafka/eagle/web/controller/ConsumersController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,7 @@ public void consumerTableAjax(HttpServletResponse response, HttpServletRequest r
e.printStackTrace();
}
obj.put("group", "<a class='link' group='" + group + "' href='#'>" + consumer.getString("group") + "</a>");
obj.put("topics", consumer.getInteger("topics"));
obj.put("topics", consumer.getString("topics"));
obj.put("node", consumer.getString("node"));
int activeTopics = consumer.getInteger("activeTopics");
int activeThreads = consumer.getInteger("activeThreads");
Expand Down Expand Up @@ -200,7 +200,7 @@ public void consumerTableAjax(HttpServletResponse response, HttpServletRequest r
e.printStackTrace();
}
obj.put("group", "<a class='link' group='" + group + "' href='#'>" + group + "</a>");
obj.put("topics", consumerSummary.getTopicNumbers());
obj.put("topics", consumerSummary.getTopics());
obj.put("node", consumerSummary.getCoordinator());
int activeTopics = consumerSummary.getActiveTopic();
int activeThreads = consumerSummary.getActiveThread();
Expand Down
8 changes: 4 additions & 4 deletions efak-web/src/main/java/org/smartloli/kafka/eagle/web/dao/TopicDao.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@
<resultMap type="org.smartloli.kafka.eagle.common.protocol.consumer.ConsumerSummaryInfo" id="csi">
<result property="cluster" column="cluster"/>
<result property="group" column="group"/>
<result property="topicNumbers" column="topic_number"/>
<result property="topics" column="topics"/>
<result property="coordinator" column="coordinator"/>
<result property="activeTopic" column="active_topic"/>
<result property="activeThread" column="active_thread_total"/>
Expand Down Expand Up @@ -311,10 +311,10 @@

<insert id="writeConsumerSummaryTopics" parameterType="java.util.List">
replace into
ke_consumer_group_summary(`cluster`,`group`,`topic_number`,`coordinator`,`active_topic`,`active_thread_total`)
ke_consumer_group_summary(`cluster`,`group`,`topics`,`coordinator`,`active_topic`,`active_thread_total`)
values
<foreach collection="list" item="item" index="index" separator=",">
(#{item.cluster},#{item.group},#{item.topicNumbers},#{item.coordinator},#{item.activeTopic},#{item.activeThread})
(#{item.cluster},#{item.group},#{item.topics},#{item.coordinator},#{item.activeTopic},#{item.activeThread})
</foreach>
</insert>

Expand Down Expand Up @@ -362,7 +362,7 @@
<where>
`cluster`=#{cluster}
<if test="search != null and search != ''">
and `group` like #{search}
and (`group` like #{search} or `topics` like #{search})
</if>
</where>
limit #{start},#{size}
Expand Down
3 changes: 2 additions & 1 deletion ...main/java/org/smartloli/kafka/eagle/web/quartz/shard/task/sub/MetricsConsumerSubTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import joptsimple.internal.Strings;
import org.apache.kafka.common.TopicPartition;
import org.smartloli.kafka.eagle.common.protocol.OwnerInfo;
import org.smartloli.kafka.eagle.common.protocol.bscreen.BScreenConsumerInfo;
Expand Down Expand Up @@ -126,7 +127,7 @@ private void bscreenConsumerTopicStats() {
ConsumerSummaryInfo csi = new ConsumerSummaryInfo();
csi.setCluster(clusterAlias);
csi.setGroup(group);
csi.setTopicNumbers(ownerInfo.getTopicSets().size());
csi.setTopics(Strings.join(ownerInfo.getTopicSets(), ","));
csi.setCoordinator(consumerGroup.getString("node"));
csi.setActiveTopic(getKafkaActiveTopicNumbers(clusterAlias, group, consumerServiceImpl));
csi.setActiveThread(ownerInfo.getActiveSize());
Expand Down
3 changes: 2 additions & 1 deletion efak-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/ConsumerServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import joptsimple.internal.Strings;
import org.smartloli.kafka.eagle.common.protocol.ConsumerInfo;
import org.smartloli.kafka.eagle.common.protocol.DisplayInfo;
import org.smartloli.kafka.eagle.common.protocol.OwnerInfo;
Expand Down Expand Up @@ -280,7 +281,7 @@ private String getKafkaConsumer(DisplayInfo page, String clusterAlias) {
consumer.setId(++id);
consumer.setNode(consumerGroup.getString("node"));
OwnerInfo ownerInfo = kafkaService.getKafkaActiverNotOwners(clusterAlias, group);
consumer.setTopics(ownerInfo.getTopicSets().size());
consumer.setTopics(Strings.join(ownerInfo.getTopicSets(), ","));
consumer.setActiveTopics(getKafkaActiveTopicNumbers(clusterAlias, group));
consumer.setActiveThreads(ownerInfo.getActiveSize());
kafkaConsumerPages.add(consumer);
Expand Down