+
*/}
+
+
+
+ );
+ },
+ });
+ }
return columns;
};
@@ -98,16 +117,33 @@ export const getGtoupTopicColumns = (arg?: any) => {
render: (t: number) => (t ? t.toLocaleString() : '-'),
},
];
- if (global.hasPermission && global.hasPermission(ClustersPermissionMap.CONSUMERS_RESET_OFFSET)) {
+ if (global.hasPermission) {
columns.push({
title: '操作',
dataIndex: 'desc',
key: 'desc',
- width: 150,
+ width: 200,
render: (value: any, record: any) => {
return (
- arg.resetOffset(record)}>重置Offset
+ {global.hasPermission(ClustersPermissionMap.CONSUMERS_RESET_OFFSET) ? (
+ arg.resetOffset(record)}>重置Offset
+ ) : (
+ <>
+ )}
+ {global.hasPermission(ClustersPermissionMap.GROUP_TOPIC_DELETE) ? (
+ arg.deleteOffset(record)}
+ okText="是"
+ cancelText="否"
+ >
+
+
+ ) : (
+ <>
+ )}
);
},
diff --git a/km-console/packages/layout-clusters-fe/src/pages/ConsumerGroup/index.tsx b/km-console/packages/layout-clusters-fe/src/pages/ConsumerGroup/index.tsx
index 8e0bb6fba..b54a5878a 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/ConsumerGroup/index.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/ConsumerGroup/index.tsx
@@ -58,6 +58,11 @@ const BrokerList: React.FC = (props: any) => {
genData({ pageNo: pagination.current, pageSize: pagination.pageSize, filters, sorter });
};
+ // 删除Group
+ const deleteTesk = () => {
+ genData({ pageNo: 1, pageSize: pagination.pageSize });
+ };
+
useEffect(() => {
genData({
pageNo: 1,
@@ -115,7 +120,7 @@ const BrokerList: React.FC = (props: any) => {
showHeader: false,
rowKey: 'group_list',
loading: loading,
- columns: getGroupColumns(),
+ columns: getGroupColumns(deleteTesk),
dataSource: data,
paginationProps: { ...pagination },
attrs: {
diff --git a/km-console/packages/layout-clusters-fe/src/pages/Login/index.tsx b/km-console/packages/layout-clusters-fe/src/pages/Login/index.tsx
index 998022311..f07e20334 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/Login/index.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/Login/index.tsx
@@ -13,7 +13,7 @@ const carouselList = [
Github:
- 5.8K
+ 6.8K
+ Star的的实时流处理平台
diff --git a/km-console/packages/layout-clusters-fe/src/pages/MirrorMaker2/index.less b/km-console/packages/layout-clusters-fe/src/pages/MirrorMaker2/index.less
index 826649028..d73f5625a 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/MirrorMaker2/index.less
+++ b/km-console/packages/layout-clusters-fe/src/pages/MirrorMaker2/index.less
@@ -185,7 +185,7 @@
.operate-connector-drawer-use-json {
.CodeMirror.cm-s-default {
- height: calc(100vh - 146px);
+ height: calc(100vh - 196px);
}
.dcloud-form-item {
margin-bottom: 0 !important;
diff --git a/km-console/packages/layout-clusters-fe/src/pages/MutliClusterPage/AccessCluster.tsx b/km-console/packages/layout-clusters-fe/src/pages/MutliClusterPage/AccessCluster.tsx
index 28992f47d..fec1a8a6b 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/MutliClusterPage/AccessCluster.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/MutliClusterPage/AccessCluster.tsx
@@ -522,28 +522,22 @@ const ConnectorForm = (props: {
const params = {
...values,
id: initFieldsValue?.id,
- jmxProperties: values.jmxProperties ? `{ "jmxProperties": "${values.jmxProperties}" }` : undefined,
+ jmxProperties: values.jmxProperties ? `{ "jmxPort": "${values.jmxProperties}" }` : undefined,
};
- Utils.put(api.batchConnectClusters, [params])
- .then((res) => {
- // setSelectedTabKey(undefined);
- getConnectClustersList();
- notification.success({
- message: '修改Connect集群成功',
- });
- })
- .catch((error) => {
- notification.success({
- message: '修改Connect集群失败',
- });
+ Utils.put(api.batchConnectClusters, [params]).then((res) => {
+ // setSelectedTabKey(undefined);
+ getConnectClustersList();
+ notification.success({
+ message: '修改Connect集群成功',
});
+ });
};
const onCancel = () => {
setSelectedTabKey(undefined);
try {
const jmxPortInfo = JSON.parse(initFieldsValue.jmxProperties) || {};
- form.setFieldsValue({ ...initFieldsValue, jmxProperties: jmxPortInfo.jmxProperties });
+ form.setFieldsValue({ ...initFieldsValue, jmxProperties: jmxPortInfo.jmxPort });
} catch {
form.setFieldsValue({ ...initFieldsValue });
}
@@ -552,7 +546,7 @@ const ConnectorForm = (props: {
useLayoutEffect(() => {
try {
const jmxPortInfo = JSON.parse(initFieldsValue.jmxProperties) || {};
- form.setFieldsValue({ ...initFieldsValue, jmxProperties: jmxPortInfo.jmxProperties });
+ form.setFieldsValue({ ...initFieldsValue, jmxProperties: jmxPortInfo.jmxPort });
} catch {
form.setFieldsValue({ ...initFieldsValue });
}
diff --git a/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/EditDrawer.tsx b/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/EditDrawer.tsx
index a325a9bce..1cc3a9099 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/EditDrawer.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/EditDrawer.tsx
@@ -132,16 +132,35 @@ const AddDrawer = forwardRef((_, ref) => {
form.validateFields().then((formData) => {
const submitData = [];
const { configType, principle, kafkaUser } = formData;
-
if (configType === 'custom') {
// 1. 自定义权限
- const { resourceType, resourcePatternType, aclPermissionType, aclOperation, aclClientHost } = formData;
+ // TODO: 需要和后端联调
+ const {
+ resourceType,
+ resourcePatternType,
+ aclPermissionType,
+ aclOperation,
+ aclClientHost,
+ cluster,
+ topicName,
+ topicPatternType,
+ groupName,
+ groupPatternType,
+ transactionalId,
+ transactionalIdPatternType,
+ } = formData;
submitData.push({
clusterId,
kafkaUser: principle === 'all' ? '*' : kafkaUser,
resourceType,
- resourcePatternType,
- resourceName: '*',
+ resourcePatternType: cluster
+ ? 3
+ : topicPatternType
+ ? topicPatternType
+ : groupPatternType
+ ? groupPatternType
+ : transactionalIdPatternType,
+ resourceName: cluster ? cluster : topicName ? topicName : groupName ? groupName : transactionalId,
aclPermissionType,
aclOperation,
aclClientHost,
@@ -281,6 +300,42 @@ const AddDrawer = forwardRef((_, ref) => {
{({ getFieldValue }) => {
+ const SelectFormItems = (props: { type: string }) => {
+ const { type } = props;
+ return (
+ ({
+ validator: (rule: any, value: string) => {
+ if (!value) {
+ return Promise.reject(`${type}Name 不能为空`);
+ }
+ if (type === 'topic' && getFieldValue(`${type}PatternType`) === ACL_PATTERN_TYPE['Literal']) {
+ return Utils.request(api.getTopicMetadata(clusterId as any, value)).then((res: any) => {
+ return res?.exist ? Promise.resolve() : Promise.reject('该 Topic 不存在');
+ });
+ }
+ return Promise.resolve();
+ },
+ }),
+ ]}
+ >
+ {
+ if (option?.value.includes(value)) {
+ return true;
+ }
+ return false;
+ }}
+ options={type === 'topic' ? topicMetaData : groupMetaData}
+ placeholder={`请输入 ${type}Name`}
+ />
+
+ );
+ };
const PatternTypeFormItems = (props: { type: string }) => {
const { type } = props;
const UpperCaseType = type[0].toUpperCase() + type.slice(1);
@@ -311,37 +366,43 @@ const AddDrawer = forwardRef((_, ref) => {
{({ getFieldValue }) =>
getFieldValue(`${type}Principle`) === 'special' ? (
- ({
- validator: (rule: any, value: string) => {
- if (!value) {
- return Promise.reject(`${UpperCaseType}Name 不能为空`);
- }
- if (type === 'topic' && getFieldValue(`${type}PatternType`) === ACL_PATTERN_TYPE['Literal']) {
- return Utils.request(api.getTopicMetadata(clusterId as any, value)).then((res: any) => {
- return res?.exist ? Promise.resolve() : Promise.reject('该 Topic 不存在');
- });
+ type !== 'transactionalId' ? (
+ ({
+ validator: (rule: any, value: string) => {
+ if (!value) {
+ return Promise.reject(`${UpperCaseType}Name 不能为空`);
+ }
+ if (type === 'topic' && getFieldValue(`${type}PatternType`) === ACL_PATTERN_TYPE['Literal']) {
+ return Utils.request(api.getTopicMetadata(clusterId as any, value)).then((res: any) => {
+ return res?.exist ? Promise.resolve() : Promise.reject('该 Topic 不存在');
+ });
+ }
+ return Promise.resolve();
+ },
+ }),
+ ]}
+ >
+ {
+ if (option?.value.includes(value)) {
+ return true;
}
- return Promise.resolve();
- },
- }),
- ]}
- >
- {
- if (option?.value.includes(value)) {
- return true;
- }
- return false;
- }}
- options={type === 'topic' ? topicMetaData : groupMetaData}
- placeholder={`请输入 ${type}Name`}
- />
-
+ return false;
+ }}
+ options={type === 'topic' ? topicMetaData : groupMetaData}
+ placeholder={`请输入 ${type}Name`}
+ />
+
+ ) : (
+
+
+
+ )
) : null
}
@@ -363,7 +424,7 @@ const AddDrawer = forwardRef((_, ref) => {
Deny
- {
Literal
Prefixed
-
+ */}
{
}))}
/>
+
+ {({ getFieldValue }) => {
+ const type = getFieldValue('resourceType');
+ if (type === ACL_RESOURCE_TYPE['Cluster']) {
+ //TODO需要和后端获取集群和事务接口联调
+ return (
+
+
+
+ );
+ } else if (type === ACL_RESOURCE_TYPE['TransactionalId']) {
+ return
{({ getFieldValue }) => {
form.resetFields(['aclOperation']);
diff --git a/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/index.tsx b/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/index.tsx
index 9919afdda..46befe501 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/index.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/SecurityACLs/index.tsx
@@ -14,6 +14,7 @@ import AddACLDrawer, {
RESOURCE_TO_OPERATIONS_MAP,
RESOURCE_MAP_KEYS,
} from './EditDrawer';
+import { ClustersPermissionMap } from '../CommonConfig';
import './index.less';
const { confirm } = Modal;
@@ -105,7 +106,7 @@ const SecurityACLs = (): JSX.Element => {
};
const columns = () => {
- const baseColumns = [
+ const baseColumns: any = [
{
title: 'Principal',
dataIndex: 'kafkaUser',
@@ -143,7 +144,9 @@ const SecurityACLs = (): JSX.Element => {
title: 'Host',
dataIndex: 'aclClientHost',
},
- {
+ ];
+ if (global.hasPermission && global.hasPermission(ClustersPermissionMap.SECURITY_ACL_DELETE)) {
+ baseColumns.push({
title: '操作',
dataIndex: '',
width: 120,
@@ -156,8 +159,8 @@ const SecurityACLs = (): JSX.Element => {
);
},
- },
- ];
+ });
+ }
return baseColumns;
};
@@ -238,15 +241,19 @@ const SecurityACLs = (): JSX.Element => {
-
- }
- onClick={() => editDrawerRef.current.onOpen(true, getACLs)}
- >
- 新增ACL
-
-
+ {global.hasPermission && global.hasPermission(ClustersPermissionMap.SECURITY_ACL_ADD) ? (
+
+ }
+ onClick={() => editDrawerRef.current.onOpen(true, getACLs)}
+ >
+ 新增ACL
+
+
+ ) : (
+ <>
+ )}
{pw}
+
+ void}
+ >
+
+
diff --git a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ConsumerGroupDetail.tsx b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ConsumerGroupDetail.tsx
index bd502b1a2..0183dca54 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ConsumerGroupDetail.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ConsumerGroupDetail.tsx
@@ -10,6 +10,7 @@ import { ClustersPermissionMap } from '../CommonConfig';
import ResetOffsetDrawer from './ResetOffsetDrawer';
import SwitchTab from '@src/components/SwitchTab';
import ContentWithCopy from '@src/components/CopyContent';
+import PubSub from "pubsub-js";
const { Option } = Select;
@@ -335,6 +336,11 @@ export default (props: any) => {
});
}, [visible]);
+// 订阅重置offset成功的消息
+ PubSub.subscribe('TopicDetail-ResetOffset', function(message, data){
+ getTopicGroupMetric({hashData: data});
+ })
+
useEffect(() => {
if (partitionList.length === 0) return;
getTopicGroupMetricHistory(partitionList, hashData);
diff --git a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ResetOffsetDrawer.tsx b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ResetOffsetDrawer.tsx
index ed948e8c0..25674eaa6 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ResetOffsetDrawer.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/ResetOffsetDrawer.tsx
@@ -4,6 +4,7 @@ import { useParams } from 'react-router-dom';
import EditTable from '../TestingProduce/component/EditTable';
import Api from '@src/api/index';
import moment from 'moment';
+import PubSub from "pubsub-js";
const CustomSelectResetTime = (props: { value?: string; onChange?: (val: Number | String) => void }) => {
const { value, onChange } = props;
@@ -106,6 +107,13 @@ export default (props: any) => {
message: '重置offset成功',
});
setResetOffsetVisible(false);
+ // 发布重置offset成功的消息
+ PubSub.publish('TopicDetail-ResetOffset',
+ {
+ groupName: record.groupName,
+ topicName: record.topicName
+ }
+ );
} else {
notification.error({
message: '重置offset失败',
diff --git a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/config.tsx b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/config.tsx
index aa1e25d3e..70055b2da 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/config.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/config.tsx
@@ -81,7 +81,8 @@ export const getTopicMessagesColmns = () => {
title: 'Offset',
dataIndex: 'offset',
key: 'offset',
- render: (t: number) => (t ? t.toLocaleString() : '-'),
+ sorter: true,
+ render: (t: number) => (+t || +t === 0 ? t.toLocaleString() : '-'), // TODO: 千分位展示
},
{
title: 'Timestamp',
diff --git a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/index.less b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/index.less
index f13249ab0..ea8470f81 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/index.less
+++ b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/index.less
@@ -26,6 +26,7 @@
.left {
display: flex;
+ align-items: center;
.info-box {
display: flex;
height: 36px;
diff --git a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/index.tsx b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/index.tsx
index ea009bd44..169f81769 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/index.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/index.tsx
@@ -15,9 +15,21 @@ import Replicator from './Replicator';
import './index.less';
import TopicDetailHealthCheck from '@src/components/CardBar/TopicDetailHealthCheck';
import { hashDataParse } from '@src/constants/common';
+import { useForceRefresh } from '@src/components/utils';
const { TabPane } = Tabs;
+const Reload = (props: any) => {
+ return (
+ void}
+ >
+
+
+ );
+};
+
const OperationsSlot: any = {
// eslint-disable-next-line react/display-name
// ['Partitions']: (arg: any) => {
@@ -70,17 +82,20 @@ const OperationsSlot: any = {
// eslint-disable-next-line react/display-name
['ConsumerGroups']: (arg: any) => {
return (
- ('');
const [visible, setVisible] = useState(false);
const [hashData, setHashData] = useState({});
+ const [refreshKey, forceRefresh] = useForceRefresh();
const callback = (key: any) => {
setSearchValue('');
@@ -184,7 +200,7 @@ const TopicDetail = (props: any) => {
onChange={callback}
tabBarExtraContent={
OperationsSlot[positionType] &&
- OperationsSlot[positionType]({ ...props, setSearchKeywords, setSearchValue, searchValue, positionType })
+ OperationsSlot[positionType]({ ...props, setSearchKeywords, setSearchValue, searchValue, positionType, forceRefresh })
}
destroyInactiveTabPane
>
@@ -196,7 +212,7 @@ const TopicDetail = (props: any) => {
{positionType === 'ConsumerGroups' && (
-
diff --git a/km-console/packages/layout-clusters-fe/src/pages/TopicList/index.tsx b/km-console/packages/layout-clusters-fe/src/pages/TopicList/index.tsx
index 4c3f88573..0a00fa049 100644
--- a/km-console/packages/layout-clusters-fe/src/pages/TopicList/index.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/TopicList/index.tsx
@@ -1,7 +1,22 @@
/* eslint-disable react/display-name */
import React, { useState, useEffect } from 'react';
import { useHistory, useParams } from 'react-router-dom';
-import { AppContainer, Input, ProTable, Select, Switch, Tooltip, Utils, Dropdown, Menu, Button, Divider, Tag } from 'knowdesign';
+import {
+ AppContainer,
+ Input,
+ ProTable,
+ Select,
+ Switch,
+ Tooltip,
+ Utils,
+ Dropdown,
+ Menu,
+ Button,
+ Divider,
+ Tag,
+ Popconfirm,
+ notification,
+} from 'knowdesign';
import { IconFont } from '@knowdesign/icons';
import Create from './Create';
import './index.less';
@@ -85,6 +100,21 @@ const AutoPage = (props: any) => {
setTopicListLoading(false);
});
};
+ const deleteTopicData = (record: any) => {
+ console.log(record, 'record');
+ const params = {
+ clusterId: Number(routeParams.clusterId),
+ topicName: record.topicName,
+ };
+ Utils.post(Api.deleteTopicData(), params).then((data: any) => {
+ if (data === null) {
+ notification.success({
+ message: '清除数据成功',
+ });
+ getTopicsList();
+ }
+ });
+ };
useEffect(() => {
getTopicsList();
}, [sortObj, showInternalTopics, searchKeywords, pageIndex, pageSize]);
@@ -247,7 +277,7 @@ const AutoPage = (props: any) => {
dataIndex: 'desc',
key: 'desc',
fixed: 'right',
- width: 140,
+ width: 200,
render: (value: any, record: any) => {
return (
> getAclFromKafka(Long clusterPhyId);
-
+public interface KafkaAclService extends MetaDataService {
List getKafkaAclFromDB(Long clusterPhyId);
Integer countKafkaAclFromDB(Long clusterPhyId);
@@ -17,10 +15,5 @@ public interface KafkaAclService {
Integer countResTypeAndDistinctFromDB(Long clusterPhyId, ResourceType resourceType);
Integer countKafkaUserAndDistinctFromDB(Long clusterPhyId);
-
- List getKafkaResTypeAclFromDB(Long clusterPhyId, Integer resType);
-
List getTopicAclFromDB(Long clusterPhyId, String topicName);
-
- List getGroupAclFromDB(Long clusterPhyId, String groupName);
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/OpKafkaAclService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/OpKafkaAclService.java
index 7dd59c754..f6129326b 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/OpKafkaAclService.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/OpKafkaAclService.java
@@ -3,10 +3,6 @@
import com.xiaojukeji.know.streaming.km.common.bean.entity.param.acl.ACLAtomParam;
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
import com.xiaojukeji.know.streaming.km.common.bean.po.KafkaAclPO;
-import org.apache.kafka.common.resource.ResourceType;
-
-import java.util.Date;
-import java.util.List;
public interface OpKafkaAclService {
/**
@@ -19,14 +15,5 @@ public interface OpKafkaAclService {
*/
Result deleteKafkaAcl(ACLAtomParam aclAtomParam, String operator);
- /**
- * 删除ACL
- */
- Result deleteKafkaAclByResName(ResourceType resourceType, String resourceName, String operator);
-
Result insertAndIgnoreDuplicate(KafkaAclPO kafkaAclPO);
-
- void batchUpdateAcls(Long clusterPhyId, List poList);
-
- int deleteByUpdateTimeBeforeInDB(Long clusterPhyId, Date beforeTime);
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/impl/KafkaAclServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/impl/KafkaAclServiceImpl.java
index 8f1473cda..65258044a 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/impl/KafkaAclServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/impl/KafkaAclServiceImpl.java
@@ -11,6 +11,7 @@
import com.xiaojukeji.know.streaming.km.common.bean.po.KafkaAclPO;
import com.xiaojukeji.know.streaming.km.common.constant.MsgConstant;
import com.xiaojukeji.know.streaming.km.common.constant.KafkaConstant;
+import com.xiaojukeji.know.streaming.km.common.converter.KafkaAclConverter;
import com.xiaojukeji.know.streaming.km.common.enums.cluster.ClusterAuthTypeEnum;
import com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemTypeEnum;
import com.xiaojukeji.know.streaming.km.common.exception.VCHandlerNotExistException;
@@ -18,8 +19,6 @@
import com.xiaojukeji.know.streaming.km.core.service.acl.KafkaAclService;
import com.xiaojukeji.know.streaming.km.core.service.cluster.ClusterPhyService;
import com.xiaojukeji.know.streaming.km.core.service.version.BaseKafkaVersionControlService;
-import com.xiaojukeji.know.streaming.km.core.service.version.BaseVersionControlService;
-import com.xiaojukeji.know.streaming.km.persistence.cache.LoadedClusterPhyCache;
import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminClient;
import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminZKClient;
import com.xiaojukeji.know.streaming.km.persistence.mysql.KafkaAclDAO;
@@ -36,11 +35,13 @@
import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.utils.SecurityUtils;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
import scala.jdk.javaapi.CollectionConverters;
@@ -77,18 +78,49 @@ private void init() {
}
@Override
- public Result> getAclFromKafka(Long clusterPhyId) {
- if (LoadedClusterPhyCache.getByPhyId(clusterPhyId) == null) {
- return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getClusterPhyNotExist(clusterPhyId));
- }
-
+ public Result> getDataFromKafka(ClusterPhy clusterPhy) {
try {
- return (Result>) versionControlService.doHandler(getVersionItemType(), getMethodName(clusterPhyId, ACL_GET_FROM_KAFKA), new ClusterPhyParam(clusterPhyId));
+ Result> dataResult = (Result>) versionControlService.doHandler(getVersionItemType(), getMethodName(clusterPhy.getId(), ACL_GET_FROM_KAFKA), new ClusterPhyParam(clusterPhy.getId()));
+ if (dataResult.failed()) {
+ Result.buildFromIgnoreData(dataResult);
+ }
+
+ return Result.buildSuc(dataResult.getData());
} catch (VCHandlerNotExistException e) {
return Result.buildFailure(e.getResultStatus());
}
}
+ @Override
+ public void writeToDB(Long clusterPhyId, List dataList) {
+ Map dbPOMap = this.getKafkaAclFromDB(clusterPhyId).stream().collect(Collectors.toMap(KafkaAclPO::getUniqueField, Function.identity()));
+
+ long now = System.currentTimeMillis();
+ for (AclBinding aclBinding: dataList) {
+ KafkaAclPO newPO = KafkaAclConverter.convert2KafkaAclPO(clusterPhyId, aclBinding, now);
+ KafkaAclPO oldPO = dbPOMap.remove(newPO.getUniqueField());
+ if (oldPO == null) {
+ // 新增的ACL
+ this.insertAndIgnoreDuplicate(newPO);
+ }
+
+ // 不需要update
+ }
+
+ // 删除已经不存在的
+ for (KafkaAclPO dbPO: dbPOMap.values()) {
+ kafkaAclDAO.deleteById(dbPO);
+ }
+ }
+
+ @Override
+ public int deleteInDBByKafkaClusterId(Long clusterPhyId) {
+ LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
+ lambdaQueryWrapper.eq(KafkaAclPO::getClusterPhyId, clusterPhyId);
+
+ return kafkaAclDAO.delete(lambdaQueryWrapper);
+ }
+
@Override
public List getKafkaAclFromDB(Long clusterPhyId) {
LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>();
@@ -116,7 +148,7 @@ public Integer countResTypeAndDistinctFromDB(Long clusterPhyId, ResourceType res
return 0;
}
- return (int)poList.stream().map(elem -> elem.getResourceName()).distinct().count();
+ return (int)poList.stream().map(KafkaAclPO::getResourceName).distinct().count();
}
@Override
@@ -130,15 +162,7 @@ public Integer countKafkaUserAndDistinctFromDB(Long clusterPhyId) {
return 0;
}
- return (int)poList.stream().map(elem -> elem.getPrincipal()).distinct().count();
- }
-
- @Override
- public List getKafkaResTypeAclFromDB(Long clusterPhyId, Integer resType) {
- LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>();
- queryWrapper.eq(KafkaAclPO::getClusterPhyId, clusterPhyId);
- queryWrapper.eq(KafkaAclPO::getResourceType, resType);
- return kafkaAclDAO.selectList(queryWrapper);
+ return (int)poList.stream().map(KafkaAclPO::getPrincipal).distinct().count();
}
@Override
@@ -152,15 +176,6 @@ public List getTopicAclFromDB(Long clusterPhyId, String topicName) {
return kafkaAclDAO.selectList(queryWrapper);
}
- @Override
- public List getGroupAclFromDB(Long clusterPhyId, String groupName) {
- LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>();
- queryWrapper.eq(KafkaAclPO::getClusterPhyId, clusterPhyId);
- queryWrapper.eq(KafkaAclPO::getResourceType, ResourceType.GROUP.code());
- queryWrapper.eq(KafkaAclPO::getResourceName, groupName);
- return kafkaAclDAO.selectList(queryWrapper);
- }
-
/**************************************************** private method ****************************************************/
private Result> getAclByZKClient(VersionItemParam itemParam){
@@ -170,7 +185,7 @@ private Result> getAclByZKClient(VersionItemParam itemParam){
for (ZkAclStore store: CollectionConverters.asJava(ZkAclStore.stores())) {
Result> rl = this.getSpecifiedTypeAclByZKClient(param.getClusterPhyId(), store.patternType());
if (rl.failed()) {
- return rl;
+ return Result.buildFromIgnoreData(rl);
}
aclList.addAll(rl.getData());
@@ -229,4 +244,19 @@ private Result> getSpecifiedTypeAclByZKClient(Long clusterPhyId
return Result.buildSuc(kafkaAclList);
}
+
+ private Result insertAndIgnoreDuplicate(KafkaAclPO kafkaAclPO) {
+ try {
+ kafkaAclDAO.insert(kafkaAclPO);
+
+ return Result.buildSuc();
+ } catch (DuplicateKeyException dke) {
+ // 直接写入，如果出现key冲突则直接忽略，因为key冲突时，表示该数据已完整存在，不需要替换任何数据
+ return Result.buildSuc();
+ } catch (Exception e) {
+ log.error("method=insertAndIgnoreDuplicate||kafkaAclPO={}||errMsg=exception", kafkaAclPO, e);
+
+ return Result.buildFromRSAndMsg(ResultStatus.MYSQL_OPERATE_FAILED, e.getMessage());
+ }
+ }
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/impl/OpKafkaAclServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/impl/OpKafkaAclServiceImpl.java
index a8fab1f1b..c3915cded 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/impl/OpKafkaAclServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/impl/OpKafkaAclServiceImpl.java
@@ -20,7 +20,6 @@
import com.xiaojukeji.know.streaming.km.core.service.acl.OpKafkaAclService;
import com.xiaojukeji.know.streaming.km.core.service.oprecord.OpLogWrapService;
import com.xiaojukeji.know.streaming.km.core.service.version.BaseKafkaVersionControlService;
-import com.xiaojukeji.know.streaming.km.core.service.version.BaseVersionControlService;
import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminClient;
import com.xiaojukeji.know.streaming.km.persistence.kafka.KafkaAdminZKClient;
import com.xiaojukeji.know.streaming.km.persistence.mysql.KafkaAclDAO;
@@ -32,7 +31,6 @@
import org.apache.kafka.common.acl.*;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourcePatternFilter;
-import org.apache.kafka.common.resource.ResourceType;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
@@ -41,8 +39,6 @@
import javax.annotation.PostConstruct;
import java.util.*;
-import java.util.function.Function;
-import java.util.stream.Collectors;
import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionEnum.*;
@@ -169,11 +165,6 @@ public Result deleteKafkaAcl(ACLAtomParam aclAtomParam, String operator) {
return rv;
}
- @Override
- public Result deleteKafkaAclByResName(ResourceType resourceType, String resourceName, String operator) {
- return Result.buildSuc();
- }
-
@Override
public Result insertAndIgnoreDuplicate(KafkaAclPO kafkaAclPO) {
try {
@@ -190,34 +181,6 @@ public Result insertAndIgnoreDuplicate(KafkaAclPO kafkaAclPO) {
}
}
- @Override
- public void batchUpdateAcls(Long clusterPhyId, List poList) {
- LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
- lambdaQueryWrapper.eq(KafkaAclPO::getClusterPhyId, clusterPhyId);
-
- Map dbPOMap = kafkaAclDAO.selectList(lambdaQueryWrapper).stream().collect(Collectors.toMap(KafkaAclPO::getUniqueField, Function.identity()));
- for (KafkaAclPO po: poList) {
- KafkaAclPO dbPO = dbPOMap.remove(po.getUniqueField());
- if (dbPO == null) {
- // 新增的ACL
- this.insertAndIgnoreDuplicate(po);
- }
- }
-
- // 删除已经不存在的
- for (KafkaAclPO dbPO: dbPOMap.values()) {
- kafkaAclDAO.deleteById(dbPO);
- }
- }
-
- @Override
- public int deleteByUpdateTimeBeforeInDB(Long clusterPhyId, Date beforeTime) {
- LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
- lambdaQueryWrapper.eq(KafkaAclPO::getClusterPhyId, clusterPhyId);
- lambdaQueryWrapper.le(KafkaAclPO::getUpdateTime, beforeTime);
- return kafkaAclDAO.delete(lambdaQueryWrapper);
- }
-
/**************************************************** private method ****************************************************/
private Result deleteInDB(KafkaAclPO kafkaAclPO) {
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/cluster/impl/ClusterPhyServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/cluster/impl/ClusterPhyServiceImpl.java
index 43a0557cb..6f9c5cfa1 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/cluster/impl/ClusterPhyServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/cluster/impl/ClusterPhyServiceImpl.java
@@ -8,6 +8,7 @@
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.ResultStatus;
import com.xiaojukeji.know.streaming.km.common.bean.event.cluster.ClusterPhyAddedEvent;
+import com.xiaojukeji.know.streaming.km.common.bean.event.cluster.connect.ClusterPhyDeletedEvent;
import com.xiaojukeji.know.streaming.km.common.bean.po.cluster.ClusterPhyPO;
import com.xiaojukeji.know.streaming.km.common.component.SpringTool;
import com.xiaojukeji.know.streaming.km.common.constant.MsgConstant;
@@ -146,6 +147,9 @@ public Result removeClusterPhyById(Long clusterPhyId, String operator) {
String.format("删除集群:%s",clusterPhy.toString()));
opLogWrapService.saveOplogAndIgnoreException(oplogDTO);
+ // 发布删除集群事件
+ SpringTool.publish(new ClusterPhyDeletedEvent(this, clusterPhyId));
+
return Result.buildSuc();
} catch (Exception e) {
log.error("method=removeClusterPhyById||clusterPhyId={}||operator={}||msg=remove cluster failed||errMsg=exception!",
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/cluster/ConnectClusterService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/cluster/ConnectClusterService.java
index e6ad39297..621a56430 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/cluster/ConnectClusterService.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/cluster/ConnectClusterService.java
@@ -4,14 +4,16 @@
import com.xiaojukeji.know.streaming.km.common.bean.dto.connect.cluster.ConnectClusterDTO;
import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.ConnectCluster;
import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.ConnectClusterMetadata;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.kafka.KSGroupDescription;
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
+import com.xiaojukeji.know.streaming.km.core.service.meta.MetaDataService;
import java.util.List;
/**
* Connect-Cluster
*/
-public interface ConnectClusterService {
+public interface ConnectClusterService extends MetaDataService {
Long replaceAndReturnIdInDB(ConnectClusterMetadata metadata);
List listByKafkaCluster(Long kafkaClusterPhyId);
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/cluster/impl/ConnectClusterMetricServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/cluster/impl/ConnectClusterMetricServiceImpl.java
index 5ed5af645..1444d3ac1 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/cluster/impl/ConnectClusterMetricServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/cluster/impl/ConnectClusterMetricServiceImpl.java
@@ -24,9 +24,9 @@
import com.xiaojukeji.know.streaming.km.core.service.connect.cluster.ConnectClusterMetricService;
import com.xiaojukeji.know.streaming.km.core.service.connect.cluster.ConnectClusterService;
import com.xiaojukeji.know.streaming.km.core.service.connect.worker.WorkerService;
-import com.xiaojukeji.know.streaming.km.core.service.version.BaseConnectorMetricService;
+import com.xiaojukeji.know.streaming.km.core.service.version.BaseConnectMetricService;
import com.xiaojukeji.know.streaming.km.persistence.connect.ConnectJMXClient;
-import com.xiaojukeji.know.streaming.km.persistence.es.dao.connect.ConnectClusterMetricESDAO;
+import com.xiaojukeji.know.streaming.km.persistence.es.dao.connect.cluster.ConnectClusterMetricESDAO;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
@@ -43,7 +43,7 @@
* @author didi
*/
@Service
-public class ConnectClusterMetricServiceImpl extends BaseConnectorMetricService implements ConnectClusterMetricService {
+public class ConnectClusterMetricServiceImpl extends BaseConnectMetricService implements ConnectClusterMetricService {
protected static final ILog LOGGER = LogFactory.getLog(ConnectClusterMetricServiceImpl.class);
public static final String CONNECT_CLUSTER_METHOD_GET_WORKER_METRIC_AVG = "getWorkerMetricAvg";
@@ -86,8 +86,7 @@ public Result collectConnectClusterMetricsFromKafkaWithCa
String connectClusterMetricKey = CollectedMetricsLocalCache.genConnectClusterMetricCacheKey(connectClusterPhyId, metric);
Float keyValue = CollectedMetricsLocalCache.getConnectClusterMetrics(connectClusterMetricKey);
if (keyValue != null) {
- ConnectClusterMetrics connectClusterMetrics = ConnectClusterMetrics.initWithMetric(connectClusterPhyId,metric,keyValue);
- return Result.buildSuc(connectClusterMetrics);
+ return Result.buildSuc(new ConnectClusterMetrics(connectClusterPhyId, metric, keyValue));
}
Result ret = this.collectConnectClusterMetricsFromKafka(connectClusterPhyId, metric);
@@ -209,8 +208,7 @@ private Result getConnectWorkerMetricByJMX(Long connectClu
try {
//2、获取jmx指标
String value = jmxConnectorWrap.getAttribute(new ObjectName(jmxInfo.getJmxObjectName()), jmxInfo.getJmxAttribute()).toString();
- ConnectWorkerMetrics connectWorkerMetrics = ConnectWorkerMetrics.initWithMetric(connectClusterId, workerId, metric, Float.valueOf(value));
- return Result.buildSuc(connectWorkerMetrics);
+ return Result.buildSuc(new ConnectWorkerMetrics(connectClusterId, workerId, metric, Float.valueOf(value)));
} catch (Exception e) {
LOGGER.error("method=getConnectWorkerMetricsByJMX||connectClusterId={}||workerId={}||metrics={}||jmx={}||msg={}",
connectClusterId, workerId, metric, jmxInfo.getJmxObjectName(), e.getClass().getName());
@@ -231,8 +229,8 @@ private List listTopNConnectClusterIdList(Long clusterPhyId, Integer topN)
.collect(Collectors.toList());
}
- protected List metricMap2VO(Long connectClusterId,
- Map>> map){
+ private List metricMap2VO(Long connectClusterId,
+ Map>> map){
List multiLinesVOS = new ArrayList<>();
if (map == null || map.isEmpty()) {
// 如果为空，则直接返回
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/cluster/impl/ConnectClusterServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/cluster/impl/ConnectClusterServiceImpl.java
index 86879662d..c0908b333 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/cluster/impl/ConnectClusterServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/cluster/impl/ConnectClusterServiceImpl.java
@@ -38,6 +38,14 @@ public class ConnectClusterServiceImpl implements ConnectClusterService {
@Autowired
private OpLogWrapService opLogWrapService;
+ @Override
+ public int deleteInDBByKafkaClusterId(Long clusterPhyId) {
+ LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
+ lambdaQueryWrapper.eq(ConnectClusterPO::getKafkaClusterPhyId, clusterPhyId);
+
+ return connectClusterDAO.deleteById(lambdaQueryWrapper);
+ }
+
@Override
public Long replaceAndReturnIdInDB(ConnectClusterMetadata metadata) {
ConnectClusterPO oldPO = this.getPOFromDB(metadata.getKafkaClusterPhyId(), metadata.getGroupName());
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/ConnectorService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/ConnectorService.java
index 220e4e895..05fe2cf95 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/ConnectorService.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/ConnectorService.java
@@ -4,49 +4,30 @@
import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.connector.KSConnector;
import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.connector.KSConnectorInfo;
import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.connector.KSConnectorStateInfo;
+import com.xiaojukeji.know.streaming.km.core.service.meta.MetaDataService;
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
import com.xiaojukeji.know.streaming.km.common.bean.po.connect.ConnectorPO;
import com.xiaojukeji.know.streaming.km.common.enums.connect.ConnectorTypeEnum;
import java.util.List;
-import java.util.Properties;
-import java.util.Set;
/**
* 查看Connector
*/
-public interface ConnectorService {
- Result createConnector(Long connectClusterId, String connectorName, Properties configs, String operator);
-
+public interface ConnectorService extends MetaDataService {
/**
* 获取所有的连接器名称列表
*/
- Result> listConnectorsFromCluster(Long connectClusterId);
+ Result> listConnectorsFromCluster(ConnectCluster connectCluster);
/**
* 获取单个连接器信息
*/
Result getConnectorInfoFromCluster(Long connectClusterId, String connectorName);
- Result> getConnectorTopicsFromCluster(Long connectClusterId, String connectorName);
-
Result getConnectorStateInfoFromCluster(Long connectClusterId, String connectorName);
- Result getAllConnectorInfoFromCluster(Long connectClusterId, String connectorName);
-
- Result resumeConnector(Long connectClusterId, String connectorName, String operator);
-
- Result restartConnector(Long connectClusterId, String connectorName, String operator);
-
- Result stopConnector(Long connectClusterId, String connectorName, String operator);
-
- Result deleteConnector(Long connectClusterId, String connectorName, String operator);
-
- Result updateConnectorConfig(Long connectClusterId, String connectorName, Properties configs, String operator);
-
- void batchReplace(Long kafkaClusterPhyId, Long connectClusterId, List connectorList, Set allConnectorNameSet);
-
- void addNewToDB(KSConnector connector);
+ Result getConnectorFromKafka(Long connectClusterId, String connectorName);
List listByKafkaClusterIdFromDB(Long kafkaClusterPhyId);
@@ -57,6 +38,4 @@ public interface ConnectorService {
ConnectorPO getConnectorFromDB(Long connectClusterId, String connectorName);
ConnectorTypeEnum getConnectorType(Long connectClusterId, String connectorName);
-
- void completeMirrorMakerInfo(ConnectCluster connectCluster, List connectorList);
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/OpConnectorService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/OpConnectorService.java
new file mode 100644
index 000000000..f94c7c080
--- /dev/null
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/OpConnectorService.java
@@ -0,0 +1,26 @@
+package com.xiaojukeji.know.streaming.km.core.service.connect.connector;
+
+import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.connector.KSConnector;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.connector.KSConnectorInfo;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
+
+import java.util.Properties;
+
+/**
+ * 查看Connector
+ */
+public interface OpConnectorService {
+ Result createConnector(Long connectClusterId, String connectorName, Properties configs, String operator);
+
+ Result resumeConnector(Long connectClusterId, String connectorName, String operator);
+
+ Result restartConnector(Long connectClusterId, String connectorName, String operator);
+
+ Result stopConnector(Long connectClusterId, String connectorName, String operator);
+
+ Result deleteConnector(Long connectClusterId, String connectorName, String operator);
+
+ Result updateConnectorConfig(Long connectClusterId, String connectorName, Properties configs, String operator);
+
+ void addNewToDB(KSConnector connector);
+}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/impl/ConnectorMetricServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/impl/ConnectorMetricServiceImpl.java
index 8792875d3..ffcc16ab4 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/impl/ConnectorMetricServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/impl/ConnectorMetricServiceImpl.java
@@ -18,6 +18,7 @@
import com.xiaojukeji.know.streaming.km.common.bean.vo.metrics.line.MetricLineVO;
import com.xiaojukeji.know.streaming.km.common.bean.vo.metrics.line.MetricMultiLinesVO;
import com.xiaojukeji.know.streaming.km.common.bean.vo.metrics.point.MetricPointVO;
+import com.xiaojukeji.know.streaming.km.common.enums.connect.ConnectStatusEnum;
import com.xiaojukeji.know.streaming.km.common.enums.connect.ConnectorTypeEnum;
import com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemTypeEnum;
import com.xiaojukeji.know.streaming.km.common.exception.VCHandlerNotExistException;
@@ -32,7 +33,7 @@
import com.xiaojukeji.know.streaming.km.core.service.connect.worker.WorkerConnectorService;
import com.xiaojukeji.know.streaming.km.core.service.connect.worker.WorkerService;
import com.xiaojukeji.know.streaming.km.core.service.health.state.HealthStateService;
-import com.xiaojukeji.know.streaming.km.core.service.version.BaseConnectorMetricService;
+import com.xiaojukeji.know.streaming.km.core.service.version.BaseConnectMetricService;
import com.xiaojukeji.know.streaming.km.persistence.connect.ConnectJMXClient;
import com.xiaojukeji.know.streaming.km.persistence.es.dao.connect.connector.ConnectorMetricESDAO;
import org.springframework.beans.factory.annotation.Autowired;
@@ -52,7 +53,7 @@
* @author didi
*/
@Service
-public class ConnectorMetricServiceImpl extends BaseConnectorMetricService implements ConnectorMetricService {
+public class ConnectorMetricServiceImpl extends BaseConnectMetricService implements ConnectorMetricService {
protected static final ILog LOGGER = LogFactory.getLog(ConnectorMetricServiceImpl.class);
public static final String CONNECTOR_METHOD_DO_NOTHING = "doNothing";
@@ -67,6 +68,8 @@ public class ConnectorMetricServiceImpl extends BaseConnectorMetricService imple
public static final String CONNECTOR_METHOD_GET_METRIC_HEALTH_SCORE = "getMetricHealthScore";
+ public static final String CONNECTOR_METHOD_GET_METRIC_RUNNING_STATUS = "getMetricRunningStatus";
+
@Autowired
private ConnectorMetricESDAO connectorMetricESDAO;
@@ -98,11 +101,12 @@ protected List listMetricPOFields() {
@Override
protected void initRegisterVCHandler() {
registerVCHandler(CONNECTOR_METHOD_DO_NOTHING, this::doNothing);
- registerVCHandler(CONNECTOR_METHOD_GET_CONNECT_WORKER_METRIC_SUM, this::getConnectWorkerMetricSum);
- registerVCHandler(CONNECTOR_METHOD_GET_CONNECTOR_TASK_METRICS_AVG, this::getConnectorTaskMetricsAvg);
- registerVCHandler(CONNECTOR_METHOD_GET_CONNECTOR_TASK_METRICS_MAX, this::getConnectorTaskMetricsMax);
- registerVCHandler(CONNECTOR_METHOD_GET_CONNECTOR_TASK_METRICS_SUM, this::getConnectorTaskMetricsSum);
- registerVCHandler(CONNECTOR_METHOD_GET_METRIC_HEALTH_SCORE, this::getMetricHealthScore);
+ registerVCHandler(CONNECTOR_METHOD_GET_CONNECT_WORKER_METRIC_SUM, this::getConnectWorkerMetricSum);
+ registerVCHandler(CONNECTOR_METHOD_GET_CONNECTOR_TASK_METRICS_AVG, this::getConnectorTaskMetricsAvg);
+ registerVCHandler(CONNECTOR_METHOD_GET_CONNECTOR_TASK_METRICS_MAX, this::getConnectorTaskMetricsMax);
+ registerVCHandler(CONNECTOR_METHOD_GET_CONNECTOR_TASK_METRICS_SUM, this::getConnectorTaskMetricsSum);
+ registerVCHandler(CONNECTOR_METHOD_GET_METRIC_HEALTH_SCORE, this::getMetricHealthScore);
+ registerVCHandler(CONNECTOR_METHOD_GET_METRIC_RUNNING_STATUS, this::getMetricRunningStatus);
}
@Override
@@ -111,8 +115,7 @@ public Result collectConnectClusterMetricsFromKafkaWithCacheFi
Float keyValue = CollectedMetricsLocalCache.getConnectorMetrics(connectorMetricKey);
if (null != keyValue) {
- ConnectorMetrics connectorMetrics = ConnectorMetrics.initWithMetric(connectClusterPhyId, connectorName, metric, keyValue);
- return Result.buildSuc(connectorMetrics);
+ return Result.buildSuc(new ConnectorMetrics(connectClusterPhyId, connectorName, metric, keyValue));
}
Result ret = this.collectConnectClusterMetricsFromKafka(connectClusterPhyId, connectorName, metric);
@@ -216,6 +219,20 @@ private Result getMetricHealthScore(VersionItemParam metricPar
return Result.buildSuc(metrics);
}
+ private Result getMetricRunningStatus(VersionItemParam metricParam) {
+ ConnectorMetricParam param = (ConnectorMetricParam) metricParam;
+ Long connectClusterId = param.getConnectClusterId();
+ String connectorName = param.getConnectorName();
+ String metricName = param.getMetricName();
+
+ ConnectorPO connector = connectorService.getConnectorFromDB(connectClusterId, connectorName);
+ if (connector == null) {
+ return Result.buildSuc(new ConnectorMetrics(connectClusterId, connectorName, metricName, (float)ConnectStatusEnum.UNKNOWN.getStatus()));
+ }
+
+ return Result.buildSuc(new ConnectorMetrics(connectClusterId, connectorName, metricName, (float)ConnectStatusEnum.getByValue(connector.getState()).getStatus()));
+ }
+
private Result getConnectWorkerMetricSum(VersionItemParam metricParam) {
ConnectorMetricParam param = (ConnectorMetricParam) metricParam;
Long connectClusterId = param.getConnectClusterId();
@@ -240,12 +257,16 @@ private Result getConnectWorkerMetricSum(VersionItemParam metr
if (!isCollected) {
return Result.buildFailure(NOT_EXIST);
}
- return Result.buildSuc(ConnectorMetrics.initWithMetric(connectClusterId, connectorName, metric, sum));
+
+ return Result.buildSuc(new ConnectorMetrics(connectClusterId, connectorName, metric, sum));
}
//kafka.connect:type=connect-worker-metrics,connector="{connector}" 指标
private Result getConnectorMetric(Long connectClusterId, String workerId, String connectorName, String metric, ConnectorTypeEnum connectorType) {
VersionConnectJmxInfo jmxInfo = getJMXInfo(connectClusterId, metric);
+ if (null == jmxInfo) {
+ return Result.buildFailure(VC_ITEM_JMX_NOT_EXIST);
+ }
if (jmxInfo.getType() != null) {
if (connectorType == null) {
@@ -257,9 +278,6 @@ private Result getConnectorMetric(Long connectClusterId, Strin
}
}
- if (null == jmxInfo) {
- return Result.buildFailure(VC_ITEM_JMX_NOT_EXIST);
- }
String jmxObjectName = String.format(jmxInfo.getJmxObjectName(), connectorName);
JmxConnectorWrap jmxConnectorWrap = connectJMXClient.getClientWithCheck(connectClusterId, workerId);
@@ -270,8 +288,7 @@ private Result getConnectorMetric(Long connectClusterId, Strin
try {
//2、获取jmx指标
String value = jmxConnectorWrap.getAttribute(new ObjectName(jmxObjectName), jmxInfo.getJmxAttribute()).toString();
- ConnectorMetrics connectorMetrics = ConnectorMetrics.initWithMetric(connectClusterId, connectorName, metric, Float.valueOf(value));
- return Result.buildSuc(connectorMetrics);
+ return Result.buildSuc(new ConnectorMetrics(connectClusterId, connectorName, metric, Float.valueOf(value)));
} catch (InstanceNotFoundException e) {
// 忽略该错误，该错误出现的原因是该指标在JMX中不存在
return Result.buildSuc(new ConnectorMetrics(connectClusterId, connectorName));
@@ -296,8 +313,7 @@ private Result getConnectorTaskMetricsAvg(VersionItemParam met
}
Float sum = ret.getData().stream().map(elem -> elem.getMetric(metric)).reduce(Float::sum).get();
- ConnectorMetrics connectorMetrics = ConnectorMetrics.initWithMetric(connectClusterId, connectorName, metric, sum / ret.getData().size());
- return Result.buildSuc(connectorMetrics);
+ return Result.buildSuc(new ConnectorMetrics(connectClusterId, connectorName, metric, sum / ret.getData().size()));
}
private Result getConnectorTaskMetricsMax(VersionItemParam metricParam){
@@ -313,8 +329,7 @@ private Result getConnectorTaskMetricsMax(VersionItemParam met
}
Float max = ret.getData().stream().max((a, b) -> a.getMetric(metric).compareTo(b.getMetric(metric))).get().getMetric(metric);
- ConnectorMetrics connectorMetrics = ConnectorMetrics.initWithMetric(connectClusterId, connectorName, metric, max);
- return Result.buildSuc(connectorMetrics);
+ return Result.buildSuc(new ConnectorMetrics(connectClusterId, connectorName, metric, max));
}
private Result getConnectorTaskMetricsSum(VersionItemParam metricParam){
@@ -330,8 +345,7 @@ private Result getConnectorTaskMetricsSum(VersionItemParam met
}
Float sum = ret.getData().stream().map(elem -> elem.getMetric(metric)).reduce(Float::sum).get();
- ConnectorMetrics connectorMetrics = ConnectorMetrics.initWithMetric(connectClusterId, connectorName, metric, sum);
- return Result.buildSuc(connectorMetrics);
+ return Result.buildSuc(new ConnectorMetrics(connectClusterId, connectorName, metric, sum));
}
@@ -358,6 +372,9 @@ private Result> getConnectorTaskMetricList(Long conne
private Result getConnectorTaskMetric(Long connectClusterId, String workerId, String connectorName, Integer taskId, String metric, ConnectorTypeEnum connectorType) {
VersionConnectJmxInfo jmxInfo = getJMXInfo(connectClusterId, metric);
+ if (null == jmxInfo) {
+ return Result.buildFailure(VC_ITEM_JMX_NOT_EXIST);
+ }
if (jmxInfo.getType() != null) {
if (connectorType == null) {
@@ -369,9 +386,6 @@ private Result getConnectorTaskMetric(Long connectClusterI
}
}
- if (null == jmxInfo) {
- return Result.buildFailure(VC_ITEM_JMX_NOT_EXIST);
- }
String jmxObjectName=String.format(jmxInfo.getJmxObjectName(), connectorName, taskId);
JmxConnectorWrap jmxConnectorWrap = connectJMXClient.getClientWithCheck(connectClusterId, workerId);
@@ -382,8 +396,7 @@ private Result getConnectorTaskMetric(Long connectClusterI
try {
//2、获取jmx指标
String value = jmxConnectorWrap.getAttribute(new ObjectName(jmxObjectName), jmxInfo.getJmxAttribute()).toString();
- ConnectorTaskMetrics connectorTaskMetrics = ConnectorTaskMetrics.initWithMetric(connectClusterId, connectorName, taskId, metric, Float.valueOf(value));
- return Result.buildSuc(connectorTaskMetrics);
+ return Result.buildSuc(new ConnectorTaskMetrics(connectClusterId, connectorName, taskId, metric, Float.valueOf(value)));
} catch (Exception e) {
LOGGER.error("method=getConnectorTaskMetric||connectClusterId={}||workerId={}||connectorName={}||taskId={}||metrics={}||jmx={}||msg={}",
connectClusterId, workerId, connectorName, taskId, metric, jmxObjectName, e.getClass().getName());
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/impl/ConnectorServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/impl/ConnectorServiceImpl.java
index 133355a84..74c298b50 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/impl/ConnectorServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/impl/ConnectorServiceImpl.java
@@ -3,7 +3,6 @@
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.didiglobal.logi.log.ILog;
import com.didiglobal.logi.log.LogFactory;
-import com.didiglobal.logi.security.common.dto.oplog.OplogDTO;
import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.ConnectCluster;
import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.connector.KSConnector;
import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.connector.KSConnectorInfo;
@@ -13,19 +12,14 @@
import com.xiaojukeji.know.streaming.km.common.bean.po.connect.ConnectorPO;
import com.xiaojukeji.know.streaming.km.common.component.RestTool;
import com.xiaojukeji.know.streaming.km.common.constant.MsgConstant;
-import com.xiaojukeji.know.streaming.km.common.constant.connect.KafkaConnectConstant;
import com.xiaojukeji.know.streaming.km.common.converter.ConnectConverter;
import com.xiaojukeji.know.streaming.km.common.enums.connect.ConnectorTypeEnum;
-import com.xiaojukeji.know.streaming.km.common.enums.operaterecord.ModuleEnum;
-import com.xiaojukeji.know.streaming.km.common.enums.operaterecord.OperationEnum;
-import com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemTypeEnum;
-import com.xiaojukeji.know.streaming.km.common.utils.BackoffUtils;
import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
+import com.xiaojukeji.know.streaming.km.common.utils.Triple;
+import com.xiaojukeji.know.streaming.km.common.utils.Tuple;
import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
import com.xiaojukeji.know.streaming.km.core.service.connect.cluster.ConnectClusterService;
import com.xiaojukeji.know.streaming.km.core.service.connect.connector.ConnectorService;
-import com.xiaojukeji.know.streaming.km.core.service.oprecord.OpLogWrapService;
-import com.xiaojukeji.know.streaming.km.core.service.version.BaseVersionControlService;
import com.xiaojukeji.know.streaming.km.persistence.mysql.connect.ConnectorDAO;
import org.apache.kafka.connect.runtime.rest.entities.ActiveTopicsInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo;
@@ -34,14 +28,9 @@
import org.springframework.stereotype.Service;
import java.util.*;
-import java.util.stream.Collectors;
-
-import static com.xiaojukeji.know.streaming.km.common.constant.connect.KafkaConnectConstant.MIRROR_MAKER_SOURCE_CLUSTER_BOOTSTRAP_SERVERS_FIELD_NAME;
-import static com.xiaojukeji.know.streaming.km.common.constant.connect.KafkaConnectConstant.MIRROR_MAKER_TARGET_CLUSTER_BOOTSTRAP_SERVERS_FIELD_NAME;
-import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemTypeEnum.SERVICE_OP_CONNECT_CONNECTOR;
@Service
-public class ConnectorServiceImpl extends BaseVersionControlService implements ConnectorService {
+public class ConnectorServiceImpl implements ConnectorService {
private static final ILog LOGGER = LogFactory.getLog(ConnectorServiceImpl.class);
@Autowired
@@ -53,79 +42,14 @@ public class ConnectorServiceImpl extends BaseVersionControlService implements C
@Autowired
private ConnectClusterService connectClusterService;
- @Autowired
- private OpLogWrapService opLogWrapService;
-
private static final String LIST_CONNECTORS_URI = "/connectors";
private static final String GET_CONNECTOR_INFO_PREFIX_URI = "/connectors";
private static final String GET_CONNECTOR_TOPICS_URI = "/connectors/%s/topics";
private static final String GET_CONNECTOR_STATUS_URI = "/connectors/%s/status";
- private static final String CREATE_CONNECTOR_URI = "/connectors";
- private static final String RESUME_CONNECTOR_URI = "/connectors/%s/resume";
- private static final String RESTART_CONNECTOR_URI = "/connectors/%s/restart";
- private static final String PAUSE_CONNECTOR_URI = "/connectors/%s/pause";
- private static final String DELETE_CONNECTOR_URI = "/connectors/%s";
- private static final String UPDATE_CONNECTOR_CONFIG_URI = "/connectors/%s/config";
-
@Override
- protected VersionItemTypeEnum getVersionItemType() {
- return SERVICE_OP_CONNECT_CONNECTOR;
- }
-
- @Override
- public Result createConnector(Long connectClusterId, String connectorName, Properties configs, String operator) {
+ public Result> listConnectorsFromCluster(ConnectCluster connectCluster) {
try {
- ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
- if (ValidateUtils.isNull(connectCluster)) {
- return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
- }
-
- // 构造参数
- Properties props = new Properties();
- props.put(KafkaConnectConstant.MIRROR_MAKER_NAME_FIELD_NAME, connectorName);
- props.put("config", configs);
-
- ConnectorInfo connectorInfo = restTool.postObjectWithJsonContent(
- connectCluster.getSuitableRequestUrl() + CREATE_CONNECTOR_URI,
- props,
- ConnectorInfo.class
- );
-
- opLogWrapService.saveOplogAndIgnoreException(new OplogDTO(
- operator,
- OperationEnum.ADD.getDesc(),
- ModuleEnum.KAFKA_CONNECT_CONNECTOR.getDesc(),
- MsgConstant.getConnectorBizStr(connectClusterId, connectorName),
- ConvertUtil.obj2Json(configs)
- ));
-
- KSConnectorInfo connector = new KSConnectorInfo();
- connector.setConnectClusterId(connectClusterId);
- connector.setConfig(connectorInfo.config());
- connector.setName(connectorInfo.name());
- connector.setTasks(connectorInfo.tasks());
- connector.setType(connectorInfo.type());
-
- return Result.buildSuc(connector);
- } catch (Exception e) {
- LOGGER.error(
- "method=createConnector||connectClusterId={}||connectorName={}||configs={}||operator={}||errMsg=exception",
- connectClusterId, connectorName, configs, operator, e
- );
-
- return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
- }
- }
-
- @Override
- public Result> listConnectorsFromCluster(Long connectClusterId) {
- try {
- ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
- if (ValidateUtils.isNull(connectCluster)) {
- return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
- }
-
List nameList = restTool.getArrayObjectWithJsonContent(
connectCluster.getSuitableRequestUrl() + LIST_CONNECTORS_URI,
new HashMap<>(),
@@ -135,8 +59,8 @@ public Result> listConnectorsFromCluster(Long connectClusterId) {
return Result.buildSuc(nameList);
} catch (Exception e) {
LOGGER.error(
- "method=listConnectorsFromCluster||connectClusterId={}||errMsg=exception",
- connectClusterId, e
+ "method=listConnectorsFromCluster||connectClusterId={}||connectClusterSuitableUrl={}||errMsg=exception",
+ connectCluster.getId(), connectCluster.getSuitableRequestUrl(), e
);
return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
@@ -153,16 +77,6 @@ public Result getConnectorInfoFromCluster(Long connectClusterId
return this.getConnectorInfoFromCluster(connectCluster, connectorName);
}
- @Override
- public Result> getConnectorTopicsFromCluster(Long connectClusterId, String connectorName) {
- ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
- if (ValidateUtils.isNull(connectCluster)) {
- return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
- }
-
- return this.getConnectorTopicsFromCluster(connectCluster, connectorName);
- }
-
@Override
public Result getConnectorStateInfoFromCluster(Long connectClusterId, String connectorName) {
ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
@@ -174,270 +88,26 @@ public Result getConnectorStateInfoFromCluster(Long connec
}
@Override
- public Result getAllConnectorInfoFromCluster(Long connectClusterId, String connectorName) {
+ public Result getConnectorFromKafka(Long connectClusterId, String connectorName) {
ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
if (ValidateUtils.isNull(connectCluster)) {
return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
}
- Result connectorResult = this.getConnectorInfoFromCluster(connectCluster, connectorName);
- if (connectorResult.failed()) {
- LOGGER.error(
- "method=getAllConnectorInfoFromCluster||connectClusterId={}||connectorName={}||result={}",
- connectClusterId, connectorName, connectorResult
- );
-
- return Result.buildFromIgnoreData(connectorResult);
- }
-
- Result> topicNameListResult = this.getConnectorTopicsFromCluster(connectCluster, connectorName);
- if (topicNameListResult.failed()) {
- LOGGER.error(
- "method=getAllConnectorInfoFromCluster||connectClusterId={}||connectorName={}||result={}",
- connectClusterId, connectorName, connectorResult
- );
- }
-
- Result stateInfoResult = this.getConnectorStateInfoFromCluster(connectCluster, connectorName);
- if (stateInfoResult.failed()) {
- LOGGER.error(
- "method=getAllConnectorInfoFromCluster||connectClusterId={}||connectorName={}||result={}",
- connectClusterId, connectorName, connectorResult
- );
+ Result, KSConnectorStateInfo>> fullInfoResult = this.getConnectorFullInfoFromKafka(connectCluster, connectorName);
+ if (fullInfoResult.failed()) {
+ return Result.buildFromIgnoreData(fullInfoResult);
}
return Result.buildSuc(ConnectConverter.convert2KSConnector(
connectCluster.getKafkaClusterPhyId(),
connectCluster.getId(),
- connectorResult.getData(),
- stateInfoResult.getData(),
- topicNameListResult.getData()
+ fullInfoResult.getData().v1(),
+ fullInfoResult.getData().v3(),
+ fullInfoResult.getData().v2()
));
}
- @Override
- public Result resumeConnector(Long connectClusterId, String connectorName, String operator) {
- try {
- ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
- if (ValidateUtils.isNull(connectCluster)) {
- return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
- }
-
- restTool.putJsonForObject(
- connectCluster.getSuitableRequestUrl() + String.format(RESUME_CONNECTOR_URI, connectorName),
- new HashMap<>(),
- String.class
- );
-
- this.updateStatus(connectCluster, connectClusterId, connectorName);
-
- opLogWrapService.saveOplogAndIgnoreException(new OplogDTO(
- operator,
- OperationEnum.ENABLE.getDesc(),
- ModuleEnum.KAFKA_CONNECT_CONNECTOR.getDesc(),
- MsgConstant.getConnectorBizStr(connectClusterId, connectorName),
- ""
- ));
-
- return Result.buildSuc();
- } catch (Exception e) {
- LOGGER.error(
- "class=ConnectorServiceImpl||method=resumeConnector||connectClusterId={}||errMsg=exception",
- connectClusterId, e
- );
-
- return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
- }
- }
-
- @Override
- public Result restartConnector(Long connectClusterId, String connectorName, String operator) {
- try {
- ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
- if (ValidateUtils.isNull(connectCluster)) {
- return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
- }
-
- restTool.postObjectWithJsonContent(
- connectCluster.getSuitableRequestUrl() + String.format(RESTART_CONNECTOR_URI, connectorName),
- new HashMap<>(),
- String.class
- );
-
- this.updateStatus(connectCluster, connectClusterId, connectorName);
-
- opLogWrapService.saveOplogAndIgnoreException(new OplogDTO(
- operator,
- OperationEnum.RESTART.getDesc(),
- ModuleEnum.KAFKA_CONNECT_CONNECTOR.getDesc(),
- MsgConstant.getConnectorBizStr(connectClusterId, connectorName),
- ""
- ));
-
- return Result.buildSuc();
- } catch (Exception e) {
- LOGGER.error(
- "method=restartConnector||connectClusterId={}||errMsg=exception",
- connectClusterId, e
- );
-
- return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
- }
- }
-
- @Override
- public Result stopConnector(Long connectClusterId, String connectorName, String operator) {
- try {
- ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
- if (ValidateUtils.isNull(connectCluster)) {
- return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
- }
-
- restTool.putJsonForObject(
- connectCluster.getSuitableRequestUrl() + String.format(PAUSE_CONNECTOR_URI, connectorName),
- new HashMap<>(),
- String.class
- );
-
- this.updateStatus(connectCluster, connectClusterId, connectorName);
-
- opLogWrapService.saveOplogAndIgnoreException(new OplogDTO(
- operator,
- OperationEnum.DISABLE.getDesc(),
- ModuleEnum.KAFKA_CONNECT_CONNECTOR.getDesc(),
- MsgConstant.getConnectorBizStr(connectClusterId, connectorName),
- ""
- ));
-
- return Result.buildSuc();
- } catch (Exception e) {
- LOGGER.error(
- "method=stopConnector||connectClusterId={}||errMsg=exception",
- connectClusterId, e
- );
-
- return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
- }
- }
-
- @Override
- public Result deleteConnector(Long connectClusterId, String connectorName, String operator) {
- try {
- ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
- if (ValidateUtils.isNull(connectCluster)) {
- return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
- }
-
- restTool.deleteWithParamsAndHeader(
- connectCluster.getSuitableRequestUrl() + String.format(DELETE_CONNECTOR_URI, connectorName),
- new HashMap<>(),
- new HashMap<>(),
- String.class
- );
-
- opLogWrapService.saveOplogAndIgnoreException(new OplogDTO(
- operator,
- OperationEnum.DELETE.getDesc(),
- ModuleEnum.KAFKA_CONNECT_CONNECTOR.getDesc(),
- MsgConstant.getConnectorBizStr(connectClusterId, connectorName),
- ""
- ));
-
- this.deleteConnectorInDB(connectClusterId, connectorName);
-
- return Result.buildSuc();
- } catch (Exception e) {
- LOGGER.error(
- "method=deleteConnector||connectClusterId={}||errMsg=exception",
- connectClusterId, e
- );
-
- return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
- }
- }
-
- @Override
- public Result updateConnectorConfig(Long connectClusterId, String connectorName, Properties configs, String operator) {
- try {
- ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
- if (ValidateUtils.isNull(connectCluster)) {
- return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
- }
-
- ConnectorInfo connectorInfo = restTool.putJsonForObject(
- connectCluster.getSuitableRequestUrl() + String.format(UPDATE_CONNECTOR_CONFIG_URI, connectorName),
- configs,
- org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo.class
- );
-
- this.updateStatus(connectCluster, connectClusterId, connectorName);
-
- opLogWrapService.saveOplogAndIgnoreException(new OplogDTO(
- operator,
- OperationEnum.EDIT.getDesc(),
- ModuleEnum.KAFKA_CONNECT_CONNECTOR.getDesc(),
- MsgConstant.getConnectorBizStr(connectClusterId, connectorName),
- ConvertUtil.obj2Json(configs)
- ));
-
- return Result.buildSuc();
- } catch (Exception e) {
- LOGGER.error(
- "method=updateConnectorConfig||connectClusterId={}||errMsg=exception",
- connectClusterId, e
- );
-
- return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
- }
- }
-
- @Override
- public void batchReplace(Long kafkaClusterPhyId, Long connectClusterId, List connectorList, Set allConnectorNameSet) {
- List poList = this.listByConnectClusterIdFromDB(connectClusterId);
-
- Map oldPOMap = new HashMap<>();
- poList.forEach(elem -> oldPOMap.put(elem.getConnectorName(), elem));
-
- for (KSConnector connector: connectorList) {
- try {
- ConnectorPO oldPO = oldPOMap.remove(connector.getConnectorName());
- if (oldPO == null) {
- oldPO = ConvertUtil.obj2Obj(connector, ConnectorPO.class);
- connectorDAO.insert(oldPO);
- } else {
- ConnectorPO newPO = ConvertUtil.obj2Obj(connector, ConnectorPO.class);
- newPO.setId(oldPO.getId());
- connectorDAO.updateById(newPO);
- }
- } catch (DuplicateKeyException dke) {
- // ignore
- }
- }
-
- try {
- oldPOMap.values().forEach(elem -> {
- if (allConnectorNameSet.contains(elem.getConnectorName())) {
- // 当前connector还存在
- return;
- }
-
- // 当前connector不存在了，则进行删除
- connectorDAO.deleteById(elem.getId());
- });
- } catch (Exception e) {
- // ignore
- }
- }
-
- @Override
- public void addNewToDB(KSConnector connector) {
- try {
- connectorDAO.insert(ConvertUtil.obj2Obj(connector, ConnectorPO.class));
- } catch (DuplicateKeyException dke) {
- // ignore
- }
- }
-
@Override
public List listByKafkaClusterIdFromDB(Long kafkaClusterPhyId) {
LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
@@ -482,53 +152,98 @@ public ConnectorTypeEnum getConnectorType(Long connectClusterId, String connecto
}
@Override
- public void completeMirrorMakerInfo(ConnectCluster connectCluster, List connectorList) {
- List sourceConnectorList = connectorList.stream().filter(elem -> elem.getConnectorClassName().equals(KafkaConnectConstant.MIRROR_MAKER_SOURCE_CONNECTOR_TYPE)).collect(Collectors.toList());
- if (sourceConnectorList.isEmpty()) {
- return;
+ public Result, List>> getDataFromKafka(ConnectCluster connectCluster) {
+ Result> nameListResult = this.listConnectorsFromCluster(connectCluster);
+ if (nameListResult.failed()) {
+ return Result.buildFromIgnoreData(nameListResult);
+ }
+
+ // 逐个获取
+ List, KSConnectorStateInfo>> connectorFullInfoList = new ArrayList<>();
+ for (String connectorName: nameListResult.getData()) {
+ Result, KSConnectorStateInfo>> ksConnectorResult = this.getConnectorFullInfoFromKafka(connectCluster, connectorName);
+ if (ksConnectorResult.failed()) {
+ continue;
+ }
+
+ connectorFullInfoList.add(ksConnectorResult.getData());
}
- List heartBeatConnectorList = connectorList.stream().filter(elem -> elem.getConnectorClassName().equals(KafkaConnectConstant.MIRROR_MAKER_HEARTBEAT_CONNECTOR_TYPE)).collect(Collectors.toList());
- List checkpointConnectorList = connectorList.stream().filter(elem -> elem.getConnectorClassName().equals(KafkaConnectConstant.MIRROR_MAKER_CHECKPOINT_CONNECTOR_TYPE)).collect(Collectors.toList());
+ // 返回结果
+ return Result.buildSuc(new Tuple<>(
+ new HashSet<>(nameListResult.getData()),
+ ConnectConverter.convertAndSupplyMirrorMakerInfo(connectCluster, connectorFullInfoList)) // 转换并补充mm2相关信息
+ );
+ }
+
+ @Override
+ public void writeToDB(Long connectClusterId, Set fullNameSet, List dataList) {
+ List poList = this.listByConnectClusterIdFromDB(connectClusterId);
+
+ Map oldPOMap = new HashMap<>();
+ poList.forEach(elem -> oldPOMap.put(elem.getConnectorName(), elem));
+
+ for (KSConnector connector: dataList) {
+ try {
+ ConnectorPO oldPO = oldPOMap.remove(connector.getConnectorName());
+ if (oldPO == null) {
+ oldPO = ConvertUtil.obj2Obj(connector, ConnectorPO.class);
+ connectorDAO.insert(oldPO);
+ continue;
+ }
- Map heartbeatMap = this.buildMirrorMakerMap(connectCluster, heartBeatConnectorList);
- Map checkpointMap = this.buildMirrorMakerMap(connectCluster, checkpointConnectorList);
+ ConnectorPO newPO = ConvertUtil.obj2Obj(connector, ConnectorPO.class);
+ newPO.setId(oldPO.getId());
+ if (!ValidateUtils.isBlank(oldPO.getCheckpointConnectorName())
+ && ValidateUtils.isBlank(newPO.getCheckpointConnectorName())
+ && fullNameSet.contains(oldPO.getCheckpointConnectorName())) {
+ // 新的po里面没有checkpoint的信息，但是db中的数据显示有，且集群中有该connector，则保留该checkpoint数据
+ newPO.setCheckpointConnectorName(oldPO.getCheckpointConnectorName());
+ }
- for (KSConnector sourceConnector : sourceConnectorList) {
- Result ret = this.getConnectorInfoFromCluster(connectCluster, sourceConnector.getConnectorName());
+ if (!ValidateUtils.isBlank(oldPO.getHeartbeatConnectorName())
+ && ValidateUtils.isBlank(newPO.getHeartbeatConnectorName())
+ && fullNameSet.contains(oldPO.getHeartbeatConnectorName())) {
+ // 新的po里面没有checkpoint的信息，但是db中的数据显示有，且集群中有该connector，则保留该checkpoint数据
+ newPO.setHeartbeatConnectorName(oldPO.getHeartbeatConnectorName());
+ }
- if (!ret.hasData()) {
+ connectorDAO.updateById(newPO);
+ } catch (DuplicateKeyException dke) {
+ // ignore
+ } catch (Exception e) {
LOGGER.error(
- "method=completeMirrorMakerInfo||connectClusterId={}||connectorName={}||get connectorInfo fail!",
- connectCluster.getId(), sourceConnector.getConnectorName()
+ "method=writeToDB||connectClusterId={}||connectorName={}||errMsg=exception",
+ connector.getConnectClusterId(), connector.getConnectorName(), e
);
- continue;
}
- KSConnectorInfo ksConnectorInfo = ret.getData();
- String targetServers = ksConnectorInfo.getConfig().get(MIRROR_MAKER_TARGET_CLUSTER_BOOTSTRAP_SERVERS_FIELD_NAME);
- String sourceServers = ksConnectorInfo.getConfig().get(MIRROR_MAKER_SOURCE_CLUSTER_BOOTSTRAP_SERVERS_FIELD_NAME);
+ }
- if (ValidateUtils.anyBlank(targetServers, sourceServers)) {
- continue;
- }
+ try {
+ oldPOMap.values().forEach(elem -> {
+ if (fullNameSet.contains(elem.getConnectorName())) {
+ // 当前connector还存在
+ return;
+ }
- String[] targetBrokerList = getBrokerList(targetServers);
- String[] sourceBrokerList = getBrokerList(sourceServers);
- sourceConnector.setHeartbeatConnectorName(this.findBindConnector(targetBrokerList, sourceBrokerList, heartbeatMap));
- sourceConnector.setCheckpointConnectorName(this.findBindConnector(targetBrokerList, sourceBrokerList, checkpointMap));
+ // 当前connector不存在了，则进行删除
+ connectorDAO.deleteById(elem.getId());
+ });
+ } catch (Exception e) {
+ // ignore
}
-
}
- /**************************************************** private method ****************************************************/
- private int deleteConnectorInDB(Long connectClusterId, String connectorName) {
+ @Override
+ public int deleteInDBByKafkaClusterId(Long clusterPhyId) {
LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
- lambdaQueryWrapper.eq(ConnectorPO::getConnectClusterId, connectClusterId);
- lambdaQueryWrapper.eq(ConnectorPO::getConnectorName, connectorName);
+ lambdaQueryWrapper.eq(ConnectorPO::getKafkaClusterPhyId, clusterPhyId);
return connectorDAO.delete(lambdaQueryWrapper);
}
+ /**************************************************** private method ****************************************************/
+
private Result getConnectorInfoFromCluster(ConnectCluster connectCluster, String connectorName) {
try {
ConnectorInfo connectorInfo = restTool.getForObject(
@@ -594,90 +309,37 @@ private Result getConnectorStateInfoFromCluster(ConnectClu
}
}
- private void updateStatus(ConnectCluster connectCluster, Long connectClusterId, String connectorName) {
- try {
- // 延迟3秒
- BackoffUtils.backoff(2000);
-
- Result stateInfoResult = this.getConnectorStateInfoFromCluster(connectCluster, connectorName);
- if (stateInfoResult.failed()) {
- return;
- }
-
- ConnectorPO po = new ConnectorPO();
- po.setConnectClusterId(connectClusterId);
- po.setConnectorName(connectorName);
- po.setState(stateInfoResult.getData().getConnector().getState());
-
- LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
- lambdaQueryWrapper.eq(ConnectorPO::getConnectClusterId, connectClusterId);
- lambdaQueryWrapper.eq(ConnectorPO::getConnectorName, connectorName);
-
- connectorDAO.update(po, lambdaQueryWrapper);
- } catch (Exception e) {
+ private Result, KSConnectorStateInfo>> getConnectorFullInfoFromKafka(ConnectCluster connectCluster, String connectorName) {
+ Result connectorResult = this.getConnectorInfoFromCluster(connectCluster, connectorName);
+ if (connectorResult.failed()) {
LOGGER.error(
- "method=updateStatus||connectClusterId={}||connectorName={}||errMsg=exception",
- connectClusterId, connectorName, e
+ "method=getConnectorAllInfoFromKafka||connectClusterId={}||connectClusterSuitableUrl={}||result={}||errMsg=get connectors info from cluster failed",
+ connectCluster.getId(), connectCluster.getSuitableRequestUrl(), connectorResult
);
- }
- }
-
- private Map buildMirrorMakerMap(ConnectCluster connectCluster, List ksConnectorList) {
- Map bindMap = new HashMap<>();
-
- for (KSConnector ksConnector : ksConnectorList) {
- Result ret = this.getConnectorInfoFromCluster(connectCluster, ksConnector.getConnectorName());
-
- if (!ret.hasData()) {
- LOGGER.error(
- "method=buildMirrorMakerMap||connectClusterId={}||connectorName={}||get connectorInfo fail!",
- connectCluster.getId(), ksConnector.getConnectorName()
- );
- continue;
- }
-
- KSConnectorInfo ksConnectorInfo = ret.getData();
- String targetServers = ksConnectorInfo.getConfig().get(MIRROR_MAKER_TARGET_CLUSTER_BOOTSTRAP_SERVERS_FIELD_NAME);
- String sourceServers = ksConnectorInfo.getConfig().get(MIRROR_MAKER_SOURCE_CLUSTER_BOOTSTRAP_SERVERS_FIELD_NAME);
-
- if (ValidateUtils.anyBlank(targetServers, sourceServers)) {
- continue;
- }
-
- String[] targetBrokerList = getBrokerList(targetServers);
- String[] sourceBrokerList = getBrokerList(sourceServers);
- for (String targetBroker : targetBrokerList) {
- for (String sourceBroker : sourceBrokerList) {
- bindMap.put(targetBroker + "@" + sourceBroker, ksConnector.getConnectorName());
- }
- }
+ return Result.buildFromIgnoreData(connectorResult);
}
- return bindMap;
- }
- private String findBindConnector(String[] targetBrokerList, String[] sourceBrokerList, Map connectorBindMap) {
- for (String targetBroker : targetBrokerList) {
- for (String sourceBroker : sourceBrokerList) {
- String connectorName = connectorBindMap.get(targetBroker + "@" + sourceBroker);
- if (connectorName != null) {
- return connectorName;
- }
- }
+ Result> topicNameListResult = this.getConnectorTopicsFromCluster(connectCluster, connectorName);
+ if (topicNameListResult.failed()) {
+ LOGGER.error(
+ "method=getConnectorAllInfoFromKafka||connectClusterId={}||connectClusterSuitableUrl={}||result={}||errMsg=get connectors topics from cluster failed",
+ connectCluster.getId(), connectCluster.getSuitableRequestUrl(), topicNameListResult
+ );
}
- return "";
- }
- private String[] getBrokerList(String str) {
- if (ValidateUtils.isBlank(str)) {
- return new String[0];
- }
- if (str.contains(";")) {
- return str.split(";");
- }
- if (str.contains(",")) {
- return str.split(",");
+ Result stateInfoResult = this.getConnectorStateInfoFromCluster(connectCluster, connectorName);
+ if (stateInfoResult.failed()) {
+ LOGGER.error(
+ "method=getConnectorAllInfoFromKafka||connectClusterId={}||connectClusterSuitableUrl={}||result={}||errMsg=get connectors state from cluster failed",
+ connectCluster.getId(), connectCluster.getSuitableRequestUrl(), stateInfoResult
+ );
}
- return new String[]{str};
+
+ return Result.buildSuc(new Triple<>(
+ connectorResult.getData(),
+ topicNameListResult.getData(),
+ stateInfoResult.getData()
+ ));
}
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/impl/OpConnectorServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/impl/OpConnectorServiceImpl.java
new file mode 100644
index 000000000..df0e96330
--- /dev/null
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/connector/impl/OpConnectorServiceImpl.java
@@ -0,0 +1,352 @@
+package com.xiaojukeji.know.streaming.km.core.service.connect.connector.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.didiglobal.logi.log.ILog;
+import com.didiglobal.logi.log.LogFactory;
+import com.didiglobal.logi.security.common.dto.oplog.OplogDTO;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.ConnectCluster;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.connector.KSConnector;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.connector.KSConnectorInfo;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.connector.KSConnectorStateInfo;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.result.ResultStatus;
+import com.xiaojukeji.know.streaming.km.common.bean.po.connect.ConnectorPO;
+import com.xiaojukeji.know.streaming.km.common.component.RestTool;
+import com.xiaojukeji.know.streaming.km.common.constant.MsgConstant;
+import com.xiaojukeji.know.streaming.km.common.constant.connect.KafkaConnectConstant;
+import com.xiaojukeji.know.streaming.km.common.enums.operaterecord.ModuleEnum;
+import com.xiaojukeji.know.streaming.km.common.enums.operaterecord.OperationEnum;
+import com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemTypeEnum;
+import com.xiaojukeji.know.streaming.km.common.utils.BackoffUtils;
+import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
+import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
+import com.xiaojukeji.know.streaming.km.core.service.connect.cluster.ConnectClusterService;
+import com.xiaojukeji.know.streaming.km.core.service.connect.connector.OpConnectorService;
+import com.xiaojukeji.know.streaming.km.core.service.oprecord.OpLogWrapService;
+import com.xiaojukeji.know.streaming.km.core.service.version.BaseVersionControlService;
+import com.xiaojukeji.know.streaming.km.persistence.mysql.connect.ConnectorDAO;
+import org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Service;
+
+import java.util.*;
+
+import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemTypeEnum.SERVICE_OP_CONNECT_CONNECTOR;
+
+@Service
+public class OpConnectorServiceImpl extends BaseVersionControlService implements OpConnectorService {
+ private static final ILog LOGGER = LogFactory.getLog(OpConnectorServiceImpl.class);
+
+ @Autowired
+ private RestTool restTool;
+
+ @Autowired
+ private ConnectorDAO connectorDAO;
+
+ @Autowired
+ private ConnectClusterService connectClusterService;
+
+ @Autowired
+ private OpLogWrapService opLogWrapService;
+
+ private static final String GET_CONNECTOR_STATUS_URI = "/connectors/%s/status";
+
+ private static final String CREATE_CONNECTOR_URI = "/connectors";
+ private static final String RESUME_CONNECTOR_URI = "/connectors/%s/resume";
+ private static final String RESTART_CONNECTOR_URI = "/connectors/%s/restart";
+ private static final String PAUSE_CONNECTOR_URI = "/connectors/%s/pause";
+ private static final String DELETE_CONNECTOR_URI = "/connectors/%s";
+ private static final String UPDATE_CONNECTOR_CONFIG_URI = "/connectors/%s/config";
+
+ @Override
+ protected VersionItemTypeEnum getVersionItemType() {
+ return SERVICE_OP_CONNECT_CONNECTOR;
+ }
+
+ @Override
+ public Result createConnector(Long connectClusterId, String connectorName, Properties configs, String operator) {
+ try {
+ ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
+ if (ValidateUtils.isNull(connectCluster)) {
+ return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
+ }
+
+ // 构造参数
+ Properties props = new Properties();
+ props.put(KafkaConnectConstant.MIRROR_MAKER_NAME_FIELD_NAME, connectorName);
+ props.put("config", configs);
+
+ ConnectorInfo connectorInfo = restTool.postObjectWithJsonContent(
+ connectCluster.getSuitableRequestUrl() + CREATE_CONNECTOR_URI,
+ props,
+ ConnectorInfo.class
+ );
+
+ opLogWrapService.saveOplogAndIgnoreException(new OplogDTO(
+ operator,
+ OperationEnum.ADD.getDesc(),
+ ModuleEnum.KAFKA_CONNECT_CONNECTOR.getDesc(),
+ MsgConstant.getConnectorBizStr(connectClusterId, connectorName),
+ ConvertUtil.obj2Json(configs)
+ ));
+
+ KSConnectorInfo connector = new KSConnectorInfo();
+ connector.setConnectClusterId(connectClusterId);
+ connector.setConfig(connectorInfo.config());
+ connector.setName(connectorInfo.name());
+ connector.setTasks(connectorInfo.tasks());
+ connector.setType(connectorInfo.type());
+
+ return Result.buildSuc(connector);
+ } catch (Exception e) {
+ LOGGER.error(
+ "method=createConnector||connectClusterId={}||connectorName={}||configs={}||operator={}||errMsg=exception",
+ connectClusterId, connectorName, configs, operator, e
+ );
+
+ return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
+ }
+ }
+
+ @Override
+ public Result resumeConnector(Long connectClusterId, String connectorName, String operator) {
+ try {
+ ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
+ if (ValidateUtils.isNull(connectCluster)) {
+ return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
+ }
+
+ restTool.putJsonForObject(
+ connectCluster.getSuitableRequestUrl() + String.format(RESUME_CONNECTOR_URI, connectorName),
+ new HashMap<>(),
+ String.class
+ );
+
+ this.updateStatus(connectCluster, connectClusterId, connectorName);
+
+ opLogWrapService.saveOplogAndIgnoreException(new OplogDTO(
+ operator,
+ OperationEnum.ENABLE.getDesc(),
+ ModuleEnum.KAFKA_CONNECT_CONNECTOR.getDesc(),
+ MsgConstant.getConnectorBizStr(connectClusterId, connectorName),
+ ""
+ ));
+
+ return Result.buildSuc();
+ } catch (Exception e) {
+ LOGGER.error(
+ "class=ConnectorServiceImpl||method=resumeConnector||connectClusterId={}||errMsg=exception",
+ connectClusterId, e
+ );
+
+ return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
+ }
+ }
+
+ @Override
+ public Result restartConnector(Long connectClusterId, String connectorName, String operator) {
+ try {
+ ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
+ if (ValidateUtils.isNull(connectCluster)) {
+ return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
+ }
+
+ restTool.postObjectWithJsonContent(
+ connectCluster.getSuitableRequestUrl() + String.format(RESTART_CONNECTOR_URI, connectorName),
+ new HashMap<>(),
+ String.class
+ );
+
+ this.updateStatus(connectCluster, connectClusterId, connectorName);
+
+ opLogWrapService.saveOplogAndIgnoreException(new OplogDTO(
+ operator,
+ OperationEnum.RESTART.getDesc(),
+ ModuleEnum.KAFKA_CONNECT_CONNECTOR.getDesc(),
+ MsgConstant.getConnectorBizStr(connectClusterId, connectorName),
+ ""
+ ));
+
+ return Result.buildSuc();
+ } catch (Exception e) {
+ LOGGER.error(
+ "method=restartConnector||connectClusterId={}||errMsg=exception",
+ connectClusterId, e
+ );
+
+ return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
+ }
+ }
+
+ @Override
+ public Result stopConnector(Long connectClusterId, String connectorName, String operator) {
+ try {
+ ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
+ if (ValidateUtils.isNull(connectCluster)) {
+ return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
+ }
+
+ restTool.putJsonForObject(
+ connectCluster.getSuitableRequestUrl() + String.format(PAUSE_CONNECTOR_URI, connectorName),
+ new HashMap<>(),
+ String.class
+ );
+
+ this.updateStatus(connectCluster, connectClusterId, connectorName);
+
+ opLogWrapService.saveOplogAndIgnoreException(new OplogDTO(
+ operator,
+ OperationEnum.DISABLE.getDesc(),
+ ModuleEnum.KAFKA_CONNECT_CONNECTOR.getDesc(),
+ MsgConstant.getConnectorBizStr(connectClusterId, connectorName),
+ ""
+ ));
+
+ return Result.buildSuc();
+ } catch (Exception e) {
+ LOGGER.error(
+ "method=stopConnector||connectClusterId={}||errMsg=exception",
+ connectClusterId, e
+ );
+
+ return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
+ }
+ }
+
+ @Override
+ public Result deleteConnector(Long connectClusterId, String connectorName, String operator) {
+ try {
+ ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
+ if (ValidateUtils.isNull(connectCluster)) {
+ return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
+ }
+
+ restTool.deleteWithParamsAndHeader(
+ connectCluster.getSuitableRequestUrl() + String.format(DELETE_CONNECTOR_URI, connectorName),
+ new HashMap<>(),
+ new HashMap<>(),
+ String.class
+ );
+
+ opLogWrapService.saveOplogAndIgnoreException(new OplogDTO(
+ operator,
+ OperationEnum.DELETE.getDesc(),
+ ModuleEnum.KAFKA_CONNECT_CONNECTOR.getDesc(),
+ MsgConstant.getConnectorBizStr(connectClusterId, connectorName),
+ ""
+ ));
+
+ this.deleteConnectorInDB(connectClusterId, connectorName);
+
+ return Result.buildSuc();
+ } catch (Exception e) {
+ LOGGER.error(
+ "method=deleteConnector||connectClusterId={}||errMsg=exception",
+ connectClusterId, e
+ );
+
+ return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
+ }
+ }
+
+ @Override
+ public Result updateConnectorConfig(Long connectClusterId, String connectorName, Properties configs, String operator) {
+ try {
+ ConnectCluster connectCluster = connectClusterService.getById(connectClusterId);
+ if (ValidateUtils.isNull(connectCluster)) {
+ return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getConnectClusterNotExist(connectClusterId));
+ }
+
+ ConnectorInfo connectorInfo = restTool.putJsonForObject(
+ connectCluster.getSuitableRequestUrl() + String.format(UPDATE_CONNECTOR_CONFIG_URI, connectorName),
+ configs,
+ ConnectorInfo.class
+ );
+
+ this.updateStatus(connectCluster, connectClusterId, connectorName);
+
+ opLogWrapService.saveOplogAndIgnoreException(new OplogDTO(
+ operator,
+ OperationEnum.EDIT.getDesc(),
+ ModuleEnum.KAFKA_CONNECT_CONNECTOR.getDesc(),
+ MsgConstant.getConnectorBizStr(connectClusterId, connectorName),
+ ConvertUtil.obj2Json(configs)
+ ));
+
+ return Result.buildSuc();
+ } catch (Exception e) {
+ LOGGER.error(
+ "method=updateConnectorConfig||connectClusterId={}||errMsg=exception",
+ connectClusterId, e
+ );
+
+ return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
+ }
+ }
+
+ @Override
+ public void addNewToDB(KSConnector connector) {
+ try {
+ connectorDAO.insert(ConvertUtil.obj2Obj(connector, ConnectorPO.class));
+ } catch (DuplicateKeyException dke) {
+ // ignore
+ }
+ }
+
+ /**************************************************** private method ****************************************************/
+ private int deleteConnectorInDB(Long connectClusterId, String connectorName) {
+ LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
+ lambdaQueryWrapper.eq(ConnectorPO::getConnectClusterId, connectClusterId);
+ lambdaQueryWrapper.eq(ConnectorPO::getConnectorName, connectorName);
+
+ return connectorDAO.delete(lambdaQueryWrapper);
+ }
+
+ private Result getConnectorStateInfoFromCluster(ConnectCluster connectCluster, String connectorName) {
+ try {
+ KSConnectorStateInfo connectorStateInfo = restTool.getForObject(
+ connectCluster.getSuitableRequestUrl() + String.format(GET_CONNECTOR_STATUS_URI, connectorName),
+ new HashMap<>(),
+ KSConnectorStateInfo.class
+ );
+
+ return Result.buildSuc(connectorStateInfo);
+ } catch (Exception e) {
+ LOGGER.error(
+ "method=getConnectorStateInfoFromCluster||connectClusterId={}||connectorName={}||errMsg=exception",
+ connectCluster.getId(), connectorName, e
+ );
+
+ return Result.buildFromRSAndMsg(ResultStatus.KAFKA_CONNECTOR_READ_FAILED, e.getMessage());
+ }
+ }
+
+ private void updateStatus(ConnectCluster connectCluster, Long connectClusterId, String connectorName) {
+ try {
+ // 延迟3秒
+ BackoffUtils.backoff(2000);
+
+ Result stateInfoResult = this.getConnectorStateInfoFromCluster(connectCluster, connectorName);
+ if (stateInfoResult.failed()) {
+ return;
+ }
+
+ ConnectorPO po = new ConnectorPO();
+ po.setConnectClusterId(connectClusterId);
+ po.setConnectorName(connectorName);
+ po.setState(stateInfoResult.getData().getConnector().getState());
+
+ LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>();
+ lambdaQueryWrapper.eq(ConnectorPO::getConnectClusterId, connectClusterId);
+ lambdaQueryWrapper.eq(ConnectorPO::getConnectorName, connectorName);
+
+ connectorDAO.update(po, lambdaQueryWrapper);
+ } catch (Exception e) {
+ LOGGER.error(
+ "method=updateStatus||connectClusterId={}||connectorName={}||errMsg=exception",
+ connectClusterId, connectorName, e
+ );
+ }
+ }
+}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/mm2/impl/MirrorMakerMetricServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/mm2/impl/MirrorMakerMetricServiceImpl.java
index 83242841d..2361be963 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/mm2/impl/MirrorMakerMetricServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/connect/mm2/impl/MirrorMakerMetricServiceImpl.java
@@ -27,7 +27,7 @@
import com.xiaojukeji.know.streaming.km.core.service.connect.connector.ConnectorService;
import com.xiaojukeji.know.streaming.km.core.service.connect.mm2.MirrorMakerMetricService;
import com.xiaojukeji.know.streaming.km.core.service.health.state.HealthStateService;
-import com.xiaojukeji.know.streaming.km.core.service.version.BaseConnectorMetricService;
+import com.xiaojukeji.know.streaming.km.core.service.version.BaseConnectMetricService;
import com.xiaojukeji.know.streaming.km.persistence.connect.ConnectJMXClient;
import org.springframework.beans.factory.annotation.Autowired;
import com.xiaojukeji.know.streaming.km.persistence.es.dao.connect.mm2.MirrorMakerMetricESDAO;
@@ -49,7 +49,7 @@
* @date 2022/12/15
*/
@Service
-public class MirrorMakerMetricServiceImpl extends BaseConnectorMetricService implements MirrorMakerMetricService {
+public class MirrorMakerMetricServiceImpl extends BaseConnectMetricService implements MirrorMakerMetricService {
protected static final ILog LOGGER = LogFactory.getLog(MirrorMakerMetricServiceImpl.class);
public static final String MIRROR_MAKER_METHOD_DO_NOTHING = "doNothing";
@@ -190,7 +190,7 @@ protected List metricMap2VO(Long connectClusterId,
multiLinesVO.setMetricLines(metricLines);
multiLinesVOS.add(multiLinesVO);
- }catch (Exception e){
+ } catch (Exception e){
LOGGER.error("method=metricMap2VO||connectClusterId={}||msg=exception!", connectClusterId, e);
}
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupServiceImpl.java
index 21511a96b..5bfb85baa 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupServiceImpl.java
@@ -78,6 +78,7 @@ public List listGroupsFromKafka(ClusterPhy clusterPhy) throws AdminOpera
}
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, clusterPhy.getBootstrapServers());
+ props.put(AdminClientConfig.CLIENT_ID_CONFIG, String.format("KSPartialAdminClient||clusterPhyId=%d||timestamp=%d", clusterPhy.getId(), System.currentTimeMillis()));
adminClient = KSPartialKafkaAdminClient.create(props);
KSListGroupsResult listConsumerGroupsResult = adminClient.listConsumerGroups(
@@ -178,6 +179,7 @@ public KSGroupDescription getGroupDescriptionFromKafka(ClusterPhy clusterPhy, St
}
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, clusterPhy.getBootstrapServers());
+ props.put(AdminClientConfig.CLIENT_ID_CONFIG, String.format("KSPartialAdminClient||clusterPhyId=%d||timestamp=%d", clusterPhy.getId(), System.currentTimeMillis()));
adminClient = KSPartialKafkaAdminClient.create(props);
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/meta/MetaDataService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/meta/MetaDataService.java
new file mode 100644
index 000000000..b1c34dbf6
--- /dev/null
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/meta/MetaDataService.java
@@ -0,0 +1,51 @@
+package com.xiaojukeji.know.streaming.km.core.service.meta;
+
+import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.ConnectCluster;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
+import com.xiaojukeji.know.streaming.km.common.utils.Tuple;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Kafka元信息服务接口
+ */
+public interface MetaDataService {
+ /**
+ * 从Kafka中获取数据
+ * @param connectCluster connect集群
+ * @return 全部资源列表, 成功的资源列表
+ */
+ default Result, List>> getDataFromKafka(ConnectCluster connectCluster) { return Result.buildSuc(new Tuple<>(new HashSet<>(), new ArrayList<>())); }
+
+ /**
+ * 从Kafka中获取数据
+ * @param clusterPhy kafka集群
+ * @return 全部资源集合, 成功的资源列表
+ */
+ default Result> getDataFromKafka(ClusterPhy clusterPhy) { return Result.buildSuc(new ArrayList<>()); }
+
+ /**
+ * 元信息同步至DB中
+ * @param clusterId 集群ID
+ * @param fullResSet 全部资源列表
+ * @param dataList 成功的资源列表
+ */
+ default void writeToDB(Long clusterId, Set fullResSet, List dataList) {}
+
+ /**
+ * 元信息同步至DB中
+ * @param clusterId 集群ID
+ * @param dataList 成功的资源列表
+ */
+ default void writeToDB(Long clusterId, List dataList) {}
+
+ /**
+ * 依据kafka集群ID删除数据
+ * @param clusterPhyId kafka集群ID
+ */
+ int deleteInDBByKafkaClusterId(Long clusterPhyId);
+}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/partition/impl/OpPartitionServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/partition/impl/OpPartitionServiceImpl.java
index 838ac5949..8fafdec53 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/partition/impl/OpPartitionServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/partition/impl/OpPartitionServiceImpl.java
@@ -19,6 +19,7 @@
import org.apache.kafka.clients.admin.ElectLeadersResult;
import org.apache.kafka.common.ElectionType;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ElectionNotNeededException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import scala.jdk.javaapi.CollectionConverters;
@@ -108,12 +109,17 @@ private Result preferredReplicaElectionByKafkaClient(VersionItemParam item
return Result.buildSuc();
} catch (Exception e) {
+ if(e.getCause() instanceof ElectionNotNeededException) {
+ // ignore ElectionNotNeededException
+ return Result.buildSuc();
+ }
+
LOGGER.error(
"method=preferredReplicaElectionByKafkaClient||clusterPhyId={}||errMsg=exception",
partitionParam.getClusterPhyId(), e
);
- return Result.buildFromRSAndMsg(ResultStatus.ZK_OPERATE_FAILED, e.getMessage());
+ return Result.buildFromRSAndMsg(ResultStatus.KAFKA_OPERATE_FAILED, e.getMessage());
}
}
}
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/BaseConnectorMetricService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/BaseConnectMetricService.java
similarity index 90%
rename from km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/BaseConnectorMetricService.java
rename to km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/BaseConnectMetricService.java
index febfdcf44..da424ebc3 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/BaseConnectorMetricService.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/BaseConnectMetricService.java
@@ -9,7 +9,7 @@
* @author wyb
* @date 2022/11/9
*/
-public abstract class BaseConnectorMetricService extends BaseConnectorVersionControlService{
+public abstract class BaseConnectMetricService extends BaseConnectVersionControlService {
private List metricNames = new ArrayList<>();
@PostConstruct
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/BaseConnectorVersionControlService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/BaseConnectVersionControlService.java
similarity index 95%
rename from km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/BaseConnectorVersionControlService.java
rename to km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/BaseConnectVersionControlService.java
index ced858ff6..8f4260614 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/BaseConnectorVersionControlService.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/BaseConnectVersionControlService.java
@@ -14,7 +14,7 @@
* @author wyb
* @date 2022/11/8
*/
-public abstract class BaseConnectorVersionControlService extends BaseVersionControlService {
+public abstract class BaseConnectVersionControlService extends BaseVersionControlService {
@Autowired
ConnectClusterService connectClusterService;
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/connect/ConnectorMetricVersionItems.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/connect/ConnectorMetricVersionItems.java
index 2d4aeac21..bcad6e3d4 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/connect/ConnectorMetricVersionItems.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/connect/ConnectorMetricVersionItems.java
@@ -24,6 +24,8 @@ public class ConnectorMetricVersionItems extends BaseMetricVersionMetric {
public static final String CONNECTOR_METRIC_HEALTH_STATE = "HealthState";
+ public static final String CONNECTOR_METRIC_RUNNING_STATUS = "RunningStatus";
+
public static final String CONNECTOR_METRIC_CONNECTOR_TOTAL_TASK_COUNT = "ConnectorTotalTaskCount";
public static final String CONNECTOR_METRIC_HEALTH_CHECK_PASSED = "HealthCheckPassed";
@@ -128,6 +130,9 @@ public List init() {
items.add(buildAllVersionsItem()
.name(CONNECTOR_METRIC_HEALTH_STATE).unit("0:好 1:中 2:差 3:宕机").desc("健康状态(0:好 1:中 2:差 3:宕机)").category(CATEGORY_HEALTH)
.extendMethod(CONNECTOR_METHOD_GET_METRIC_HEALTH_SCORE));
+ items.add(buildAllVersionsItem()
+ .name(CONNECTOR_METRIC_RUNNING_STATUS).unit("0:UNASSIGNED 1:RUNNING 2:PAUSED 3:FAILED 4:DESTROYED -1:UNKNOWN").desc("运行状态(0:UNASSIGNED 1:RUNNING 2:PAUSED 3:FAILED 4:DESTROYED -1:UNKNOWN)").category(CATEGORY_PERFORMANCE)
+ .extendMethod(CONNECTOR_METHOD_GET_METRIC_RUNNING_STATUS));
items.add(buildAllVersionsItem()
.name(CONNECTOR_METRIC_HEALTH_CHECK_PASSED).unit("个").desc("健康项检查通过数").category(CATEGORY_HEALTH)
.extendMethod(CONNECTOR_METHOD_GET_METRIC_HEALTH_SCORE));
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/kafka/ClusterMetricVersionItems.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/kafka/ClusterMetricVersionItems.java
index cefe8930e..dbcdac57f 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/kafka/ClusterMetricVersionItems.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/metrics/kafka/ClusterMetricVersionItems.java
@@ -299,7 +299,7 @@ public List init(){
// MessagesIn 指标
itemList.add( buildAllVersionsItem()
- .name(CLUSTER_METRIC_MESSAGES_IN).unit("条/s").desc("集群每条消息写入条数").category(CATEGORY_CLUSTER)
+ .name(CLUSTER_METRIC_MESSAGES_IN).unit("条/s").desc("集群每秒消息写入条数").category(CATEGORY_CLUSTER)
.extend( buildJMXMethodExtend( CLUSTER_METHOD_GET_METRIC_FROM_KAFKA_BY_TOTAL_BROKERS_JMX )
.jmxObjectName( JMX_SERVER_BROKER_MESSAGES_IN ).jmxAttribute(RATE_MIN_1)));
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/zookeeper/ZookeeperService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/zookeeper/ZookeeperService.java
index 8d3a78b10..1d324928f 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/zookeeper/ZookeeperService.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/zookeeper/ZookeeperService.java
@@ -1,19 +1,11 @@
package com.xiaojukeji.know.streaming.km.core.service.zookeeper;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.config.ZKConfig;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
import com.xiaojukeji.know.streaming.km.common.bean.entity.zookeeper.ZookeeperInfo;
+import com.xiaojukeji.know.streaming.km.core.service.meta.MetaDataService;
import java.util.List;
-public interface ZookeeperService {
- /**
- * 从ZK集群中获取ZK信息
- */
- Result> listFromZookeeper(Long clusterPhyId, String zookeeperAddress, ZKConfig zkConfig);
-
- void batchReplaceDataInDB(Long clusterPhyId, List infoList);
-
+public interface ZookeeperService extends MetaDataService {
List listFromDBByCluster(Long clusterPhyId);
/**
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/zookeeper/impl/ZookeeperServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/zookeeper/impl/ZookeeperServiceImpl.java
index 8b0d63d1f..dc2f58d24 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/zookeeper/impl/ZookeeperServiceImpl.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/zookeeper/impl/ZookeeperServiceImpl.java
@@ -3,6 +3,7 @@
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.didiglobal.logi.log.ILog;
import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
import com.xiaojukeji.know.streaming.km.common.bean.entity.config.ZKConfig;
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.ResultStatus;
@@ -22,10 +23,8 @@
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
+import java.util.stream.Collectors;
@Service
public class ZookeeperServiceImpl implements ZookeeperService {
@@ -35,14 +34,14 @@ public class ZookeeperServiceImpl implements ZookeeperService {
private ZookeeperDAO zookeeperDAO;
@Override
- public Result
@@ -257,6 +287,19 @@ const AutoPage = (props: any) => {
<>
)}
{global.hasPermission(ClustersPermissionMap.TOPIC_DEL) ? deleteTopicData(record)}
+ okText="是"
+ cancelText="否"
+ >
+
+
+ ) : (
+ <>
+ )}
);
},
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/KafkaAclService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/KafkaAclService.java
index 3e50771b5..9e9735a96 100644
--- a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/KafkaAclService.java
+++ b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/acl/KafkaAclService.java
@@ -1,15 +1,13 @@
package com.xiaojukeji.know.streaming.km.core.service.acl;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
import com.xiaojukeji.know.streaming.km.common.bean.po.KafkaAclPO;
+import com.xiaojukeji.know.streaming.km.core.service.meta.MetaDataService;
import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.resource.ResourceType;
import java.util.List;
-public interface KafkaAclService {
- Result