You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by do...@apache.org on 2023/01/06 13:53:02 UTC
[inlong] branch branch-1.5 updated (43b821ac4 -> e719c62b5)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a change to branch branch-1.5
in repository https://gitbox.apache.org/repos/asf/inlong.git
from 43b821ac4 [INLONG-7166][DataProxy] Fix audit data reporting (#7167)
new dabfef741 [INLONG-7161][Sort] Fix bug that Mysql connector only output the latest record in snapshot stage for table without primary key (Supplement) (#7175)
new 7cb6f77f3 [INLONG-7172][Sort] Fix new table write into iceberg failed (#7173)
new c5d98cfeb [INLONG-7169][Manager] Optimize OpenDataNodeController implementation (#7170)
new 933828234 [INLONG-7176][Dashboard] Data flow group adds restart and stop state operations (#7177)
new e719c62b5 [INLONG-7163][Dashboard] Data subscription addition and details page optimization (#7179)
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
inlong-dashboard/src/locales/cn.json | 8 +++
inlong-dashboard/src/locales/en.json | 8 +++
.../metas/consumes/common/ConsumeDefaultInfo.ts | 45 ++++++++++------
.../src/metas/consumes/defaults/Pulsar.ts | 4 +-
.../src/pages/ConsumeDetail/Info/config.tsx | 28 ++++++++++
.../src/pages/ConsumeDetail/Info/index.tsx | 11 +++-
inlong-dashboard/src/pages/ConsumeDetail/index.tsx | 4 +-
.../src/pages/GroupDashboard/config.tsx | 12 ++++-
.../src/pages/GroupDashboard/index.tsx | 36 ++++++++++++-
...teValidation.java => UpdateByIdValidation.java} | 2 +-
...eValidation.java => UpdateByKeyValidation.java} | 6 +--
.../manager/pojo/node/DataNodePageRequest.java | 6 +++
.../inlong/manager/pojo/node/DataNodeRequest.java | 16 +++---
.../pojo/node/ck/ClickHouseDataNodeDTO.java | 3 +-
.../pojo/node/es/ElasticsearchDataNodeDTO.java | 4 +-
.../manager/pojo/node/hive/HiveDataNodeDTO.java | 4 +-
.../manager/pojo/node/hudi/HudiDataNodeDTO.java | 4 +-
.../pojo/node/iceberg/IcebergDataNodeDTO.java | 4 +-
.../manager/pojo/node/mysql/MySQLDataNodeDTO.java | 3 +-
.../pojo/node/starrocks/StarRocksDataNodeDTO.java | 3 +-
.../service/node/AbstractDataNodeOperator.java | 7 +--
.../service/node/DataNodeOperatorFactory.java | 1 +
.../manager/service/node/DataNodeServiceImpl.java | 62 ++++++++++++++--------
.../node/ck/ClickHouseDataNodeOperator.java | 4 +-
.../node/es/ElasticsearchDataNodeOperator.java | 7 +--
.../service/node/hive/HiveDataNodeOperator.java | 7 +--
.../service/node/hudi/HudiDataNodeOperator.java | 6 +--
.../node/iceberg/IcebergDataNodeOperator.java | 7 +--
.../service/node/mysql/MySQLDataNodeOperator.java | 7 +--
.../node/starrocks/StarRocksDataNodeOperator.java | 7 +--
.../manager/web/controller/DataNodeController.java | 11 ++--
.../controller/openapi/OpenDataNodeController.java | 14 ++---
.../web/controller/DataNodeControllerTest.java | 1 -
.../sink/multiple/IcebergSingleFileCommiter.java | 3 +-
.../mysql/debezium/reader/BinlogSplitReader.java | 3 +-
35 files changed, 243 insertions(+), 115 deletions(-)
copy inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/validation/{UpdateValidation.java => UpdateByIdValidation.java} (95%)
copy inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/validation/{UpdateValidation.java => UpdateByKeyValidation.java} (89%)
[inlong] 04/05: [INLONG-7176][Dashboard] Data flow group adds restart and stop state operations (#7177)
Posted by do...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.5
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 933828234927489e7327d359f5213b47e3d1cc75
Author: Lizhen <88...@users.noreply.github.com>
AuthorDate: Fri Jan 6 18:55:14 2023 +0800
[INLONG-7176][Dashboard] Data flow group adds restart and stop state operations (#7177)
---
inlong-dashboard/src/locales/cn.json | 6 ++++
inlong-dashboard/src/locales/en.json | 6 ++++
.../src/pages/GroupDashboard/config.tsx | 12 +++++++-
.../src/pages/GroupDashboard/index.tsx | 36 +++++++++++++++++++++-
4 files changed, 58 insertions(+), 2 deletions(-)
diff --git a/inlong-dashboard/src/locales/cn.json b/inlong-dashboard/src/locales/cn.json
index bbb003abe..7b56eb441 100644
--- a/inlong-dashboard/src/locales/cn.json
+++ b/inlong-dashboard/src/locales/cn.json
@@ -453,9 +453,15 @@
"pages.GroupDashboard.config.WaitApproveCount": "待审批",
"pages.GroupDashboard.config.Reject": "已驳回",
"pages.GroupDashboard.config.ExecuteLog": "日志",
+ "pages.GroupDashboard.config.Restart": "重启",
+ "pages.GroupDashboard.config.Stop": "停止",
"pages.GroupDashboard.ConfirmDelete": "确认删除吗",
"pages.GroupDashboard.Create": "新建数据流组",
"pages.GroupDashboard.SuccessfullyDeleted": "删除成功",
+ "pages.GroupDashboard.ConfirmRestart": "确认重启吗",
+ "pages.GroupDashboard.SuccessfullyRestart": "重启成功",
+ "pages.GroupDashboard.ConfirmStop": "确认停止吗",
+ "pages.GroupDashboard.SuccessfullyStop": "停止成功",
"pages.GroupDetail.Info.Basic": "基础信息",
"pages.GroupDetail.Info.Mq": "MQ 信息",
"pages.GroupDetail.Info.Back": "返回",
diff --git a/inlong-dashboard/src/locales/en.json b/inlong-dashboard/src/locales/en.json
index 1b8551a20..d243632c9 100644
--- a/inlong-dashboard/src/locales/en.json
+++ b/inlong-dashboard/src/locales/en.json
@@ -453,9 +453,15 @@
"pages.GroupDashboard.config.WaitApproveCount": "WaitApproveCount",
"pages.GroupDashboard.config.Reject": "Reject",
"pages.GroupDashboard.config.ExecuteLog": "ExecuteLog",
+ "pages.GroupDashboard.config.Restart": "Restart",
+ "pages.GroupDashboard.config.Stop": "Stop",
"pages.GroupDashboard.ConfirmDelete": "Confirm delete?",
"pages.GroupDashboard.Create": "Create",
"pages.GroupDashboard.SuccessfullyDeleted": "Successfully deleted",
+ "pages.GroupDashboard.ConfirmRestart": "Confirm restart?",
+ "pages.GroupDashboard.SuccessfullyRestart": "Successfully restart",
+ "pages.GroupDashboard.ConfirmStop": "Confirm stop?",
+ "pages.GroupDashboard.SuccessfullyStop": "Successfully stop",
"pages.GroupDetail.Info.Basic": "Basic",
"pages.GroupDetail.Info.Mq": "MQ",
"pages.GroupDetail.Info.Back": "Back",
diff --git a/inlong-dashboard/src/pages/GroupDashboard/config.tsx b/inlong-dashboard/src/pages/GroupDashboard/config.tsx
index 7fb3a8dd3..049b9ca39 100644
--- a/inlong-dashboard/src/pages/GroupDashboard/config.tsx
+++ b/inlong-dashboard/src/pages/GroupDashboard/config.tsx
@@ -47,7 +47,7 @@ export const dashCardList = [
},
];
-export const useColumns = ({ onDelete, openModal }) => {
+export const useColumns = ({ onDelete, openModal, onRestart, onStop }) => {
const { defaultValue } = useDefaultMeta('group');
const { Entity } = useLoadMeta<GroupMetaType>('group', defaultValue);
@@ -86,6 +86,16 @@ export const useColumns = ({ onDelete, openModal }) => {
<Button type="link" onClick={() => onDelete(record)}>
{i18n.t('basic.Delete')}
</Button>
+ {record?.status && (record?.status === 120 || record?.status === 130) && (
+ <Button type="link" onClick={() => onRestart(record)}>
+ {i18n.t('pages.GroupDashboard.config.Restart')}
+ </Button>
+ )}
+ {record?.status && (record?.status === 120 || record?.status === 130) && (
+ <Button type="link" onClick={() => onStop(record)}>
+ {i18n.t('pages.GroupDashboard.config.Stop')}
+ </Button>
+ )}
{record?.status && (record?.status === 120 || record?.status === 130) && (
<Button type="link" onClick={() => openModal(record)}>
{i18n.t('pages.GroupDashboard.config.ExecuteLog')}
diff --git a/inlong-dashboard/src/pages/GroupDashboard/index.tsx b/inlong-dashboard/src/pages/GroupDashboard/index.tsx
index c9b896fae..04642b0e4 100644
--- a/inlong-dashboard/src/pages/GroupDashboard/index.tsx
+++ b/inlong-dashboard/src/pages/GroupDashboard/index.tsx
@@ -85,6 +85,40 @@ const Comp: React.FC = () => {
setGroupLogs({ visible: true, inlongGroupId: inlongGroupId });
};
+ const onRestart = ({ inlongGroupId }) => {
+ Modal.confirm({
+ title: i18n.t('pages.GroupDashboard.ConfirmRestart'),
+ onOk: async () => {
+ await request({
+ url: `/group/restartProcess/${inlongGroupId}`,
+ method: 'POST',
+ data: {
+ groupId: inlongGroupId,
+ },
+ });
+ await getList();
+ message.success(i18n.t('pages.GroupDashboard.SuccessfullyRestart'));
+ },
+ });
+ };
+
+ const onStop = ({ inlongGroupId }) => {
+ Modal.confirm({
+ title: i18n.t('pages.GroupDashboard.ConfirmStop'),
+ onOk: async () => {
+ await request({
+ url: `/group/suspendProcess/${inlongGroupId}`,
+ method: 'POST',
+ data: {
+ groupId: inlongGroupId,
+ },
+ });
+ await getList();
+ message.success(i18n.t('pages.GroupDashboard.SuccessfullyStop'));
+ },
+ });
+ };
+
const onChange = ({ current: pageNum, pageSize }) => {
setOptions(prev => ({
...prev,
@@ -112,7 +146,7 @@ const Comp: React.FC = () => {
title: summary[item.dataIndex] || 0,
}));
- const columns = useColumns({ onDelete, openModal });
+ const columns = useColumns({ onDelete, openModal, onRestart, onStop });
const getFilterFormContent = useCallback(
defaultValues => [
[inlong] 01/05: [INLONG-7161][Sort] Fix bug that Mysql connector only output the latest record in snapshot stage for table without primary key (Supplement) (#7175)
Posted by do...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.5
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit dabfef7410d2156d60359c9a8a0daff702168eb1
Author: Schnapps <zp...@connect.ust.hk>
AuthorDate: Fri Jan 6 16:21:28 2023 +0800
[INLONG-7161][Sort] Fix bug that Mysql connector only output the latest record in snapshot stage for table without primary key (Supplement) (#7175)
Co-authored-by: stingpeng <st...@tencent.com>
---
.../inlong/sort/cdc/mysql/debezium/reader/BinlogSplitReader.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/inlong-sort/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/reader/BinlogSplitReader.java b/inlong-sort/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/reader/BinlogSplitReader.java
index 2a455de98..2e6a9f85d 100644
--- a/inlong-sort/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/reader/BinlogSplitReader.java
+++ b/inlong-sort/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/reader/BinlogSplitReader.java
@@ -188,7 +188,8 @@ public class BinlogSplitReader implements DebeziumReader<SourceRecord, MySqlSpli
if (isDataChangeRecord(sourceRecord)) {
TableId tableId = getTableId(sourceRecord);
BinlogOffset position = getBinlogPosition(sourceRecord);
- if (hasEnterPureBinlogPhase(tableId, position)) {
+ // source record has no primary need no comparing for binlog position
+ if (hasEnterPureBinlogPhase(tableId, position) || sourceRecord.key() == null) {
return true;
}
// only the table who captured snapshot splits need to filter
[inlong] 05/05: [INLONG-7163][Dashboard] Data subscription addition and details page optimization (#7179)
Posted by do...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.5
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit e719c62b5e1387688ea4e1b2b5d11aea4afcf68e
Author: Lizhen <88...@users.noreply.github.com>
AuthorDate: Fri Jan 6 21:49:25 2023 +0800
[INLONG-7163][Dashboard] Data subscription addition and details page optimization (#7179)
---
inlong-dashboard/src/locales/cn.json | 2 +
inlong-dashboard/src/locales/en.json | 2 +
.../metas/consumes/common/ConsumeDefaultInfo.ts | 45 ++++++++++++++--------
.../src/metas/consumes/defaults/Pulsar.ts | 4 +-
.../src/pages/ConsumeDetail/Info/config.tsx | 28 ++++++++++++++
.../src/pages/ConsumeDetail/Info/index.tsx | 11 +++++-
inlong-dashboard/src/pages/ConsumeDetail/index.tsx | 4 +-
7 files changed, 76 insertions(+), 20 deletions(-)
diff --git a/inlong-dashboard/src/locales/cn.json b/inlong-dashboard/src/locales/cn.json
index 7b56eb441..e9ca2b47e 100644
--- a/inlong-dashboard/src/locales/cn.json
+++ b/inlong-dashboard/src/locales/cn.json
@@ -387,6 +387,7 @@
"meta.Consume.No": "否",
"meta.Consume.Owner": "订阅责任人",
"meta.Consume.OwnersExtra": "可查看、修改订阅信息",
+ "meta.Consume.Pulsar.isDiq": "配置死信队列",
"meta.Nodes.Name": "节点名称",
"meta.Nodes.Type": "类型",
"meta.Nodes.Owners": "责任人",
@@ -585,6 +586,7 @@
"pages.ConsumeDashboard.status.Cancelled": "已取消",
"pages.ConsumeDashboard.NewSubscribe": "新建订阅",
"pages.ConsumeDetail.SubscribeDetails": "订阅详情",
+ "pages.ConsumeDetail.ClusterInfo": "集群信息:",
"pages.Clusters.Type": "类型",
"pages.Clusters.TypeAll": "全部",
"pages.Clusters.Create": "新建集群",
diff --git a/inlong-dashboard/src/locales/en.json b/inlong-dashboard/src/locales/en.json
index d243632c9..ff0bd4de7 100644
--- a/inlong-dashboard/src/locales/en.json
+++ b/inlong-dashboard/src/locales/en.json
@@ -387,6 +387,7 @@
"meta.Consume.No": "No",
"meta.Consume.Owner": "Subscription Owners",
"meta.Consume.OwnersExtra": "Can view, modify subscription info",
+ "meta.Consume.Pulsar.isDiq": "Configure dead letter queue",
"meta.Nodes.Name": "Name",
"meta.Nodes.Type": "Type",
"meta.Nodes.Owners": "Owners",
@@ -585,6 +586,7 @@
"pages.ConsumeDashboard.status.Cancelled": "Cancelled",
"pages.ConsumeDashboard.NewSubscribe": "Create",
"pages.ConsumeDetail.SubscribeDetails": "Subscription Detail",
+ "pages.ConsumeDetail.ClusterInfo": "ClusterInfo:",
"pages.Clusters.Type": "Type",
"pages.Clusters.TypeAll": "All",
"pages.Clusters.Create": "Create",
diff --git a/inlong-dashboard/src/metas/consumes/common/ConsumeDefaultInfo.ts b/inlong-dashboard/src/metas/consumes/common/ConsumeDefaultInfo.ts
index a49c86d34..3accd2810 100644
--- a/inlong-dashboard/src/metas/consumes/common/ConsumeDefaultInfo.ts
+++ b/inlong-dashboard/src/metas/consumes/common/ConsumeDefaultInfo.ts
@@ -70,11 +70,37 @@ export class ConsumeDefaultInfo implements DataWithBackend, RenderRow, RenderLis
@I18n('meta.Consume.Owner')
inCharges: string;
+ @FieldDecorator({
+ type: 'radio',
+ rules: [{ required: true }],
+ props: {
+ options: [
+ {
+ label: 'Kafka',
+ value: 'KAFKA',
+ },
+ {
+ label: 'Pulsar',
+ value: 'PULSAR',
+ },
+ {
+ label: 'TubeMQ',
+ value: 'TUBEMQ',
+ },
+ ],
+ },
+ })
+ @ColumnDecorator({
+ render: text => consumes.find(c => c.value === text)?.label || text,
+ })
+ @I18n('meta.Consume.MQType')
+ mqType: string;
+
@FieldDecorator({
type: 'select',
extraNames: ['mqType'],
rules: [{ required: true }],
- props: {
+ props: values => ({
showSearch: true,
filterOption: false,
options: {
@@ -84,6 +110,7 @@ export class ConsumeDefaultInfo implements DataWithBackend, RenderRow, RenderLis
method: 'POST',
data: {
keyword,
+ mqType: values.mqType,
pageNum: 1,
pageSize: 20,
status: 130,
@@ -93,7 +120,7 @@ export class ConsumeDefaultInfo implements DataWithBackend, RenderRow, RenderLis
formatResult: result =>
result?.list?.map(item => ({
...item,
- label: `${item.inlongGroupId} (${item.mqType})`,
+ label: item.inlongGroupId,
value: item.inlongGroupId,
})),
},
@@ -102,18 +129,12 @@ export class ConsumeDefaultInfo implements DataWithBackend, RenderRow, RenderLis
topic: undefined,
mqType: option.mqType,
}),
- },
+ }),
})
@ColumnDecorator()
@I18n('meta.Consume.TargetInlongGroupID')
inlongGroupId: string;
- @ColumnDecorator({
- render: text => consumes.find(c => c.value === text)?.label || text,
- })
- @I18n('meta.Consume.MQType')
- mqType: string;
-
@FieldDecorator({
type: 'select',
rules: [{ required: true }],
@@ -187,12 +208,6 @@ export class ConsumeDefaultInfo implements DataWithBackend, RenderRow, RenderLis
@I18n('pages.ConsumeDashboard.config.OperatingStatus')
readonly lastConsumeStatus: string;
- @FieldDecorator({
- type: 'text',
- })
- @I18n('meta.Consume.MQAddress')
- readonly masterUrl: string;
-
parse(data) {
return data;
}
diff --git a/inlong-dashboard/src/metas/consumes/defaults/Pulsar.ts b/inlong-dashboard/src/metas/consumes/defaults/Pulsar.ts
index 04e0ac284..af13a3042 100644
--- a/inlong-dashboard/src/metas/consumes/defaults/Pulsar.ts
+++ b/inlong-dashboard/src/metas/consumes/defaults/Pulsar.ts
@@ -47,7 +47,7 @@ export default class PulsarConsume
],
},
})
- @I18n('isDlq')
+ @I18n('meta.Consume.Pulsar.isDiq')
isDlq: 0 | 1;
@FieldDecorator({
@@ -76,7 +76,7 @@ export default class PulsarConsume
},
visible: values => values?.isDlq,
})
- @I18n('isRlq')
+ @I18n('meta.Consume.Pulsar.isDiq')
isRlq: 0 | 1;
@FieldDecorator({
diff --git a/inlong-dashboard/src/pages/ConsumeDetail/Info/config.tsx b/inlong-dashboard/src/pages/ConsumeDetail/Info/config.tsx
index 858689352..e327ebc5f 100644
--- a/inlong-dashboard/src/pages/ConsumeDetail/Info/config.tsx
+++ b/inlong-dashboard/src/pages/ConsumeDetail/Info/config.tsx
@@ -20,6 +20,9 @@
import { useMemo } from 'react';
import { useLoadMeta, ConsumeMetaType } from '@/metas';
import { excludeObjectArray } from '@/utils';
+import React from 'react';
+import { Table } from 'antd';
+import i18n from '@/i18n';
export const useFormContent = ({ mqType, editing, isCreate }) => {
const { Entity } = useLoadMeta<ConsumeMetaType>('consume', mqType);
@@ -47,6 +50,31 @@ export const useFormContent = ({ mqType, editing, isCreate }) => {
}));
};
+export const getFormContent = ({ clusterInfos, isCreate, formContent }) => {
+ const array = [
+ ...formContent,
+ {
+ type: <label>{i18n.t('pages.ConsumeDetail.ClusterInfo')}</label>,
+ },
+ {
+ type: (
+ <Table
+ size="small"
+ columns={[
+ { title: 'name', dataIndex: 'name' },
+ { title: 'type', dataIndex: 'type' },
+ { title: 'serviceUrl', dataIndex: 'url' },
+ { title: 'adminUrl', dataIndex: 'adminUrl' },
+ ]}
+ dataSource={clusterInfos}
+ rowKey="name"
+ />
+ ),
+ },
+ ];
+ return isCreate ? formContent : array;
+};
+
function transType(editing: boolean, conf) {
const arr = [
{
diff --git a/inlong-dashboard/src/pages/ConsumeDetail/Info/index.tsx b/inlong-dashboard/src/pages/ConsumeDetail/Info/index.tsx
index 61d7b6a2f..9f631c272 100644
--- a/inlong-dashboard/src/pages/ConsumeDetail/Info/index.tsx
+++ b/inlong-dashboard/src/pages/ConsumeDetail/Info/index.tsx
@@ -25,7 +25,7 @@ import request from '@/utils/request';
import { useTranslation } from 'react-i18next';
import { useDefaultMeta } from '@/metas';
import { CommonInterface } from '../common';
-import { useFormContent } from './config';
+import { getFormContent, useFormContent } from './config';
type Props = CommonInterface;
@@ -37,6 +37,8 @@ const Comp = ({ id, readonly, isCreate }: Props, ref) => {
const [mqType, setMqType] = useState(defaultValue);
+ const [clusterInfos, setClusterInfos] = useState(defaultValue);
+
const [form] = useForm();
const isUpdate = useMemo(() => {
@@ -54,6 +56,7 @@ const Comp = ({ id, readonly, isCreate }: Props, ref) => {
onSuccess: data => {
form.setFieldsValue(data);
setMqType(data.mqType);
+ setClusterInfos(data.clusterInfos);
},
});
@@ -104,7 +107,11 @@ const Comp = ({ id, readonly, isCreate }: Props, ref) => {
<div style={{ position: 'relative' }}>
<FormGenerator
form={form}
- content={formContent}
+ content={getFormContent({
+ clusterInfos,
+ isCreate,
+ formContent,
+ })}
initialValues={data}
onValuesChange={(c, values) => setMqType(values.mqType)}
useMaxWidth={800}
diff --git a/inlong-dashboard/src/pages/ConsumeDetail/index.tsx b/inlong-dashboard/src/pages/ConsumeDetail/index.tsx
index dcadd70fd..4b6917b64 100644
--- a/inlong-dashboard/src/pages/ConsumeDetail/index.tsx
+++ b/inlong-dashboard/src/pages/ConsumeDetail/index.tsx
@@ -55,7 +55,9 @@ const Comp: React.FC = () => {
const list = useMemo(
() => [
{
- label: t('pages.ConsumeDetail.SubscribeDetails'),
+ label: isCreate
+ ? t('pages.ConsumeCreate.NewSubscribe')
+ : t('pages.ConsumeDetail.SubscribeDetails'),
value: 'consumeDetail',
content: Info,
},
[inlong] 02/05: [INLONG-7172][Sort] Fix new table write into iceberg failed (#7173)
Posted by do...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.5
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 7cb6f77f3d3cfc25ed7865aeab8df485b039e234
Author: emhui <11...@users.noreply.github.com>
AuthorDate: Fri Jan 6 16:23:19 2023 +0800
[INLONG-7172][Sort] Fix new table write into iceberg failed (#7173)
---
.../inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java b/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java
index 4f13c8ec6..6c5e14fe7 100644
--- a/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java
+++ b/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java
@@ -160,7 +160,8 @@ public class IcebergSingleFileCommiter extends IcebergProcessFunction<WriteResul
this.checkpointsState = context.getOperatorStateStore().getListState(stateDescriptor);
this.jobIdState = context.getOperatorStateStore().getListState(jobIdDescriptor);
- if (context.isRestored()) {
+ // New table doesn't have state, so it doesn't need to do restore operation.
+ if (context.isRestored() && jobIdState.get().iterator().hasNext()) {
String restoredFlinkJobId = jobIdState.get().iterator().next();
Preconditions.checkState(!Strings.isNullOrEmpty(restoredFlinkJobId),
"Flink job id parsed from checkpoint snapshot shouldn't be null or empty");
[inlong] 03/05: [INLONG-7169][Manager] Optimize OpenDataNodeController implementation (#7170)
Posted by do...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.5
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit c5d98cfebaeb437894e9966c1dfb29a343ec44c7
Author: Goson Zhang <46...@qq.com>
AuthorDate: Fri Jan 6 17:00:28 2023 +0800
[INLONG-7169][Manager] Optimize OpenDataNodeController implementation (#7170)
---
.../common/validation/UpdateByIdValidation.java | 36 +++++++++++++
.../common/validation/UpdateByKeyValidation.java | 36 +++++++++++++
.../manager/pojo/node/DataNodePageRequest.java | 6 +++
.../inlong/manager/pojo/node/DataNodeRequest.java | 16 +++---
.../pojo/node/ck/ClickHouseDataNodeDTO.java | 3 +-
.../pojo/node/es/ElasticsearchDataNodeDTO.java | 4 +-
.../manager/pojo/node/hive/HiveDataNodeDTO.java | 4 +-
.../manager/pojo/node/hudi/HudiDataNodeDTO.java | 4 +-
.../pojo/node/iceberg/IcebergDataNodeDTO.java | 4 +-
.../manager/pojo/node/mysql/MySQLDataNodeDTO.java | 3 +-
.../pojo/node/starrocks/StarRocksDataNodeDTO.java | 3 +-
.../service/node/AbstractDataNodeOperator.java | 7 +--
.../service/node/DataNodeOperatorFactory.java | 1 +
.../manager/service/node/DataNodeServiceImpl.java | 62 ++++++++++++++--------
.../node/ck/ClickHouseDataNodeOperator.java | 4 +-
.../node/es/ElasticsearchDataNodeOperator.java | 7 +--
.../service/node/hive/HiveDataNodeOperator.java | 7 +--
.../service/node/hudi/HudiDataNodeOperator.java | 6 +--
.../node/iceberg/IcebergDataNodeOperator.java | 7 +--
.../service/node/mysql/MySQLDataNodeOperator.java | 7 +--
.../node/starrocks/StarRocksDataNodeOperator.java | 7 +--
.../manager/web/controller/DataNodeController.java | 11 ++--
.../controller/openapi/OpenDataNodeController.java | 14 ++---
.../web/controller/DataNodeControllerTest.java | 1 -
24 files changed, 173 insertions(+), 87 deletions(-)
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/validation/UpdateByIdValidation.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/validation/UpdateByIdValidation.java
new file mode 100644
index 000000000..16714faf7
--- /dev/null
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/validation/UpdateByIdValidation.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.common.validation;
+
+import javax.validation.groups.Default;
+
+/**
+ * Used for validate update request fields group
+ *
+ * <p/>
+ * In general, the request body of save and update can be shared,
+ * but we need to verify the parameters of the two requests separately
+ *
+ * <p/>
+ * For example, the request body save and update only have the difference in id,
+ * and this id must be carried when updating, we can use it like this
+ * <code>org.apache.inlong.manager.pojo.node.DataNodeRequest</code>
+ */
+public interface UpdateByIdValidation extends Default {
+
+}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/validation/UpdateByKeyValidation.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/validation/UpdateByKeyValidation.java
new file mode 100644
index 000000000..028c0807d
--- /dev/null
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/validation/UpdateByKeyValidation.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.common.validation;
+
+import javax.validation.groups.Default;
+
+/**
+ * Used for validate update request fields group
+ *
+ * <p/>
+ * In general, the request body of save and update can be shared,
+ * but we need to verify the parameters of the two requests separately
+ *
+ * <p/>
+ * For example, the request body save and update only have the difference in keys,
+ * and this keys must be carried when updating, we can use it like this
+ * <code>org.apache.inlong.manager.pojo.node.DataNodeRequest</code>
+ */
+public interface UpdateByKeyValidation extends Default {
+
+}
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/DataNodePageRequest.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/DataNodePageRequest.java
index 496aa9c9c..21ceaa60d 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/DataNodePageRequest.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/DataNodePageRequest.java
@@ -22,6 +22,9 @@ import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import lombok.EqualsAndHashCode;
import org.apache.inlong.manager.pojo.common.PageRequest;
+import org.hibernate.validator.constraints.Length;
+
+import javax.validation.constraints.Pattern;
/**
* Data node paging query conditions
@@ -32,9 +35,12 @@ import org.apache.inlong.manager.pojo.common.PageRequest;
public class DataNodePageRequest extends PageRequest {
@ApiModelProperty(value = "Data node type, including MYSQL, HIVE, KAFKA, ES, etc.")
+ @Length(max = 20, message = "length must be less than or equal to 20")
private String type;
@ApiModelProperty(value = "Data node name")
+ @Length(min = 1, max = 128, message = "length must be between 1 and 128")
+ @Pattern(regexp = "^[A-Za-z0-9_-]{1,128}$", message = "only supports letters, numbers, '-', or '_'")
private String name;
@ApiModelProperty(value = "Keywords, name, url, etc.")
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/DataNodeRequest.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/DataNodeRequest.java
index d4c3d89cb..0172b10d4 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/DataNodeRequest.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/DataNodeRequest.java
@@ -23,7 +23,10 @@ import io.swagger.annotations.ApiModelProperty;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
-import org.apache.inlong.manager.common.validation.UpdateValidation;
+
+import org.apache.inlong.manager.common.validation.SaveValidation;
+import org.apache.inlong.manager.common.validation.UpdateByIdValidation;
+import org.apache.inlong.manager.common.validation.UpdateByKeyValidation;
import org.hibernate.validator.constraints.Length;
import javax.validation.constraints.NotBlank;
@@ -40,18 +43,18 @@ import javax.validation.constraints.Pattern;
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "type")
public abstract class DataNodeRequest {
- @NotNull(groups = UpdateValidation.class)
@ApiModelProperty(value = "Primary key")
+ @NotNull(groups = UpdateByIdValidation.class, message = "id cannot be null")
private Integer id;
- @NotBlank(message = "node name cannot be blank")
@ApiModelProperty(value = "Data node name")
+ @NotBlank(groups = {SaveValidation.class, UpdateByKeyValidation.class}, message = "node name cannot be blank")
@Length(min = 1, max = 128, message = "length must be between 1 and 128")
@Pattern(regexp = "^[A-Za-z0-9_-]{1,128}$", message = "only supports letters, numbers, '-', or '_'")
private String name;
- @NotBlank(message = "node type cannot be blank")
@ApiModelProperty(value = "Data node type, including MYSQL, HIVE, KAFKA, ES, etc.")
+ @NotBlank(message = "node type cannot be blank")
@Length(max = 20, message = "length must be less than or equal to 20")
private String type;
@@ -75,12 +78,13 @@ public abstract class DataNodeRequest {
@Length(max = 256, message = "length must be less than or equal to 256")
private String description;
- @NotBlank(message = "inCharges cannot be blank")
- @ApiModelProperty(value = "Name of responsible person, separated by commas", required = true)
+ @ApiModelProperty(value = "Name of responsible person, separated by commas")
+ @NotBlank(groups = SaveValidation.class, message = "inCharges cannot be blank")
@Length(max = 512, message = "length must be less than or equal to 512")
private String inCharges;
@ApiModelProperty(value = "Version number")
+ @NotNull(groups = {UpdateByIdValidation.class, UpdateByKeyValidation.class}, message = "version cannot be null")
private Integer version;
}
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/ck/ClickHouseDataNodeDTO.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/ck/ClickHouseDataNodeDTO.java
index 8843db2ae..c308b01df 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/ck/ClickHouseDataNodeDTO.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/ck/ClickHouseDataNodeDTO.java
@@ -48,7 +48,8 @@ public class ClickHouseDataNodeDTO {
try {
return JsonUtils.parseObject(extParams, ClickHouseDataNodeDTO.class);
} catch (Exception e) {
- throw new BusinessException(ErrorCodeEnum.GROUP_INFO_INCORRECT.getMessage());
+ throw new BusinessException(ErrorCodeEnum.GROUP_INFO_INCORRECT,
+ String.format("Failed to parse extParams for ClickHouse node: %s", e.getMessage()));
}
}
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/es/ElasticsearchDataNodeDTO.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/es/ElasticsearchDataNodeDTO.java
index 701dc248b..0fe8548a3 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/es/ElasticsearchDataNodeDTO.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/es/ElasticsearchDataNodeDTO.java
@@ -85,8 +85,8 @@ public class ElasticsearchDataNodeDTO {
try {
return JsonUtils.parseObject(extParams, ElasticsearchDataNodeDTO.class);
} catch (Exception e) {
- LOGGER.error("Failed to extract additional parameters for Elasticsearch data node: ", e);
- throw new BusinessException(ErrorCodeEnum.GROUP_INFO_INCORRECT.getMessage());
+ throw new BusinessException(ErrorCodeEnum.GROUP_INFO_INCORRECT,
+ String.format("Failed to parse extParams for Elasticsearch node: %s", e.getMessage()));
}
}
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/hive/HiveDataNodeDTO.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/hive/HiveDataNodeDTO.java
index 74b57864d..ea9d36c6b 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/hive/HiveDataNodeDTO.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/hive/HiveDataNodeDTO.java
@@ -78,8 +78,8 @@ public class HiveDataNodeDTO {
try {
return JsonUtils.parseObject(extParams, HiveDataNodeDTO.class);
} catch (Exception e) {
- LOGGER.error("Failed to extract additional parameters for hive data node: ", e);
- throw new BusinessException(ErrorCodeEnum.GROUP_INFO_INCORRECT.getMessage());
+ throw new BusinessException(ErrorCodeEnum.GROUP_INFO_INCORRECT,
+ String.format("Failed to parse extParams for Hive node: %s", e.getMessage()));
}
}
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/hudi/HudiDataNodeDTO.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/hudi/HudiDataNodeDTO.java
index 53a71e81c..68dbaf2fa 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/hudi/HudiDataNodeDTO.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/hudi/HudiDataNodeDTO.java
@@ -66,8 +66,8 @@ public class HudiDataNodeDTO {
try {
return JsonUtils.parseObject(extParams, HudiDataNodeDTO.class);
} catch (Exception e) {
- LOGGER.error("Failed to extract additional parameters for Hudi data node: ", e);
- throw new BusinessException(ErrorCodeEnum.GROUP_INFO_INCORRECT.getMessage());
+ throw new BusinessException(ErrorCodeEnum.GROUP_INFO_INCORRECT,
+ String.format("Failed to parse extParams for Hudi node: %s", e.getMessage()));
}
}
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/iceberg/IcebergDataNodeDTO.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/iceberg/IcebergDataNodeDTO.java
index d1aada9ec..9baadcf9d 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/iceberg/IcebergDataNodeDTO.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/iceberg/IcebergDataNodeDTO.java
@@ -67,8 +67,8 @@ public class IcebergDataNodeDTO {
try {
return JsonUtils.parseObject(extParams, IcebergDataNodeDTO.class);
} catch (Exception e) {
- LOGGER.error("Failed to extract additional parameters for iceberg data node: ", e);
- throw new BusinessException(ErrorCodeEnum.GROUP_INFO_INCORRECT.getMessage());
+ throw new BusinessException(ErrorCodeEnum.GROUP_INFO_INCORRECT,
+ String.format("Failed to parse extParams for Iceberg node: %s", e.getMessage()));
}
}
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/mysql/MySQLDataNodeDTO.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/mysql/MySQLDataNodeDTO.java
index 4ed05e257..633c7a67b 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/mysql/MySQLDataNodeDTO.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/mysql/MySQLDataNodeDTO.java
@@ -62,7 +62,8 @@ public class MySQLDataNodeDTO {
try {
return JsonUtils.parseObject(extParams, MySQLDataNodeDTO.class);
} catch (Exception e) {
- throw new BusinessException(ErrorCodeEnum.CLUSTER_INFO_INCORRECT.getMessage() + ": " + e.getMessage());
+ throw new BusinessException(ErrorCodeEnum.CLUSTER_INFO_INCORRECT,
+ String.format("Failed to parse extParams for MySQL node: %s", e.getMessage()));
}
}
}
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/starrocks/StarRocksDataNodeDTO.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/starrocks/StarRocksDataNodeDTO.java
index c01d99c6b..a7148baa1 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/starrocks/StarRocksDataNodeDTO.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/starrocks/StarRocksDataNodeDTO.java
@@ -57,7 +57,8 @@ public class StarRocksDataNodeDTO {
try {
return JsonUtils.parseObject(extParams, StarRocksDataNodeDTO.class);
} catch (Exception e) {
- throw new BusinessException(ErrorCodeEnum.GROUP_INFO_INCORRECT.getMessage());
+ throw new BusinessException(ErrorCodeEnum.GROUP_INFO_INCORRECT,
+ String.format("Failed to parse extParams for StarRocks node: %s", e.getMessage()));
}
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/AbstractDataNodeOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/AbstractDataNodeOperator.java
index 91fae43a8..a1aed8856 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/AbstractDataNodeOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/AbstractDataNodeOperator.java
@@ -75,9 +75,10 @@ public abstract class AbstractDataNodeOperator implements DataNodeOperator {
entity.setModifier(operator);
int rowCount = dataNodeEntityMapper.updateByIdSelective(entity);
if (rowCount != InlongConstants.AFFECTED_ONE_ROW) {
- LOGGER.error("data node has already updated with name={}, type={}, request version={}, updated row={}",
- request.getName(), request.getType(), request.getVersion(), rowCount);
- throw new BusinessException(ErrorCodeEnum.CONFIG_EXPIRED);
+ throw new BusinessException(ErrorCodeEnum.CONFIG_EXPIRED,
+ String.format(
+ "failure to update data node with name=%s, type=%s, request version=%d, updated row=%d",
+ request.getName(), request.getType(), request.getVersion(), rowCount));
}
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/DataNodeOperatorFactory.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/DataNodeOperatorFactory.java
index fba9d84d7..5761c3e44 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/DataNodeOperatorFactory.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/DataNodeOperatorFactory.java
@@ -41,6 +41,7 @@ public class DataNodeOperatorFactory {
.filter(inst -> inst.accept(type))
.findFirst()
.orElseThrow(() -> new BusinessException(
+ ErrorCodeEnum.DATA_NODE_TYPE_NOT_SUPPORTED,
String.format(ErrorCodeEnum.DATA_NODE_TYPE_NOT_SUPPORTED.getMessage(), type)));
}
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/DataNodeServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/DataNodeServiceImpl.java
index 60935ed54..614ff8ca2 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/DataNodeServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/DataNodeServiceImpl.java
@@ -19,6 +19,8 @@ package org.apache.inlong.manager.service.node;
import com.github.pagehelper.Page;
import com.github.pagehelper.PageHelper;
+
+import org.apache.commons.lang3.StringUtils;
import org.apache.inlong.manager.common.consts.InlongConstants;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.UserTypeEnum;
@@ -129,7 +131,7 @@ public class DataNodeServiceImpl implements DataNodeService {
}
DataNodeEntity entity = dataNodeMapper.selectById(id);
if (entity == null) {
- throw new BusinessException("data node not found");
+ throw new BusinessException(ErrorCodeEnum.DATA_NODE_NOT_FOUND);
}
String dataNodeType = entity.getType();
DataNodeOperator dataNodeOperator = operatorFactory.getInstance(dataNodeType);
@@ -195,18 +197,26 @@ public class DataNodeServiceImpl implements DataNodeService {
@Transactional(rollbackFor = Throwable.class)
public Boolean update(DataNodeRequest request, String operator) {
LOGGER.info("begin to update data node by id: {}", request);
-
- // Check whether the data node name exists with the same groupId and streamId
- String name = request.getName();
- String type = request.getType();
- DataNodeEntity existEntity = dataNodeMapper.selectByUniqueKey(name, type);
- Integer id = request.getId();
- if (existEntity != null && !existEntity.getId().equals(id)) {
- String errMsg = String.format("data node already exist for name=%s, type=%s", name, type);
- LOGGER.error(errMsg);
- throw new BusinessException(errMsg);
+ // check whether record existed
+ DataNodeEntity curEntity = dataNodeMapper.selectById(request.getId());
+ if (curEntity == null) {
+ throw new BusinessException(ErrorCodeEnum.RECORD_NOT_FOUND,
+ String.format("data node record not found by id=%d", request.getId()));
+ }
+ // Check whether the data node name exists with the same name and type
+ if (request.getName() != null) {
+ if (StringUtils.isBlank(request.getName())) {
+ throw new BusinessException(ErrorCodeEnum.INVALID_PARAMETER,
+ "the name changed of data node is blank!");
+ }
+ DataNodeEntity existEntity =
+ dataNodeMapper.selectByUniqueKey(request.getName(), request.getType());
+ if (existEntity != null && !existEntity.getId().equals(request.getId())) {
+ throw new BusinessException(ErrorCodeEnum.RECORD_DUPLICATE,
+ String.format("data node already exist for name=%s, type=%s, required id=%s, exist id=%s",
+ request.getName(), request.getType(), request.getId(), existEntity.getId()));
+ }
}
-
DataNodeOperator dataNodeOperator = operatorFactory.getInstance(request.getType());
dataNodeOperator.updateOpt(request, operator);
@@ -229,17 +239,25 @@ public class DataNodeServiceImpl implements DataNodeService {
if (!opInfo.getRoles().contains(UserTypeEnum.ADMIN.name())) {
throw new BusinessException(ErrorCodeEnum.PERMISSION_REQUIRED);
}
- // Check whether the data node name exists with the same groupId and streamId
- DataNodeEntity existEntity =
- dataNodeMapper.selectByUniqueKey(request.getName(), request.getType());
- if (existEntity == null) {
- throw new BusinessException(ErrorCodeEnum.DATA_NODE_NOT_FOUND);
+ // check the record existed
+ DataNodeEntity curEntity = dataNodeMapper.selectById(request.getId());
+ if (curEntity == null) {
+ throw new BusinessException(ErrorCodeEnum.RECORD_NOT_FOUND,
+ String.format("data node record not found by id=%d", request.getId()));
}
- Integer id = request.getId();
- if (id != null && !existEntity.getId().equals(id)) {
- throw new BusinessException(ErrorCodeEnum.DATA_NODE_ID_CHANGED,
- String.format("data node already exist for name=%s, type=%s, required id=%s, exist id=%s",
- request.getName(), request.getType(), id, existEntity.getId()));
+ // Check whether the data node name exists with the same name and type
+ if (request.getName() != null) {
+ if (StringUtils.isBlank(request.getName())) {
+ throw new BusinessException(ErrorCodeEnum.INVALID_PARAMETER,
+ "the name changed of data node is blank!");
+ }
+ DataNodeEntity existEntity =
+ dataNodeMapper.selectByUniqueKey(request.getName(), request.getType());
+ if (existEntity != null && !existEntity.getId().equals(request.getId())) {
+ throw new BusinessException(ErrorCodeEnum.RECORD_DUPLICATE,
+ String.format("data node already exist for name=%s, type=%s, required id=%s, exist id=%s",
+ request.getName(), request.getType(), request.getId(), existEntity.getId()));
+ }
}
DataNodeOperator dataNodeOperator = operatorFactory.getInstance(request.getType());
dataNodeOperator.updateOpt(request, opInfo.getName());
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/ck/ClickHouseDataNodeOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/ck/ClickHouseDataNodeOperator.java
index 7f386416b..5ae29f234 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/ck/ClickHouseDataNodeOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/ck/ClickHouseDataNodeOperator.java
@@ -81,8 +81,8 @@ public class ClickHouseDataNodeOperator extends AbstractDataNodeOperator {
ClickHouseDataNodeDTO dto = ClickHouseDataNodeDTO.getFromRequest(ckDataNodeRequest);
targetEntity.setExtParams(objectMapper.writeValueAsString(dto));
} catch (Exception e) {
- LOGGER.error("failed to set entity for hive data node: ", e);
- throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT.getMessage());
+ throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT,
+ String.format("Failed to build extParams for ClickHouse node: %s", e.getMessage()));
}
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/es/ElasticsearchDataNodeOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/es/ElasticsearchDataNodeOperator.java
index e07338f09..a04e4b531 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/es/ElasticsearchDataNodeOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/es/ElasticsearchDataNodeOperator.java
@@ -64,10 +64,9 @@ public class ElasticsearchDataNodeOperator extends AbstractDataNodeOperator {
try {
ElasticsearchDataNodeDTO dto = ElasticsearchDataNodeDTO.getFromRequest(esRequest);
targetEntity.setExtParams(objectMapper.writeValueAsString(dto));
- LOGGER.debug("success to set entity for elasticsearch data node");
} catch (Exception e) {
- LOGGER.error("failed to set entity for elasticsearch data node: ", e);
- throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT.getMessage());
+ throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT,
+ String.format("Failed to build extParams for Elasticsearch node: %s", e.getMessage()));
}
}
@@ -82,8 +81,6 @@ public class ElasticsearchDataNodeOperator extends AbstractDataNodeOperator {
ElasticsearchDataNodeDTO dto = ElasticsearchDataNodeDTO.getFromJson(entity.getExtParams());
CommonBeanUtils.copyProperties(dto, info);
}
-
- LOGGER.debug("success to get elasticsearch data node from entity");
return info;
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/hive/HiveDataNodeOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/hive/HiveDataNodeOperator.java
index 6adbe0b1a..d5d6c7eb8 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/hive/HiveDataNodeOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/hive/HiveDataNodeOperator.java
@@ -69,8 +69,6 @@ public class HiveDataNodeOperator extends AbstractDataNodeOperator {
HiveDataNodeDTO dto = HiveDataNodeDTO.getFromJson(entity.getExtParams());
CommonBeanUtils.copyProperties(dto, hiveDataNodeInfo);
}
-
- LOGGER.debug("success to get hive data node from entity");
return hiveDataNodeInfo;
}
@@ -81,10 +79,9 @@ public class HiveDataNodeOperator extends AbstractDataNodeOperator {
try {
HiveDataNodeDTO dto = HiveDataNodeDTO.getFromRequest(hiveDataNodeRequest);
targetEntity.setExtParams(objectMapper.writeValueAsString(dto));
- LOGGER.debug("success to set entity for hive data node");
} catch (Exception e) {
- LOGGER.error("failed to set entity for hive data node: ", e);
- throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT.getMessage());
+ throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT,
+ String.format("Failed to build extParams for Hive node: %s", e.getMessage()));
}
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/hudi/HudiDataNodeOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/hudi/HudiDataNodeOperator.java
index 79b253cd9..3d8e68cfe 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/hudi/HudiDataNodeOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/hudi/HudiDataNodeOperator.java
@@ -65,8 +65,6 @@ public class HudiDataNodeOperator extends AbstractDataNodeOperator {
HudiDataNodeDTO dto = HudiDataNodeDTO.getFromJson(entity.getExtParams());
CommonBeanUtils.copyProperties(dto, hudiDataNodeInfo);
}
-
- LOGGER.debug("success to get Hudi data node from entity");
return hudiDataNodeInfo;
}
@@ -78,8 +76,8 @@ public class HudiDataNodeOperator extends AbstractDataNodeOperator {
HudiDataNodeDTO dto = HudiDataNodeDTO.getFromRequest(hudiDataNodeRequest);
targetEntity.setExtParams(objectMapper.writeValueAsString(dto));
} catch (Exception e) {
- LOGGER.error("failed to set entity for Hudi data node: ", e);
- throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT.getMessage());
+ throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT,
+ String.format("Failed to build extParams for Hudi node: %s", e.getMessage()));
}
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/iceberg/IcebergDataNodeOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/iceberg/IcebergDataNodeOperator.java
index 74506feaa..3959bdc2b 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/iceberg/IcebergDataNodeOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/iceberg/IcebergDataNodeOperator.java
@@ -61,15 +61,12 @@ public class IcebergDataNodeOperator extends AbstractDataNodeOperator {
if (entity == null) {
throw new BusinessException(ErrorCodeEnum.DATA_NODE_NOT_FOUND);
}
-
IcebergDataNodeInfo icebergDataNodeInfo = new IcebergDataNodeInfo();
CommonBeanUtils.copyProperties(entity, icebergDataNodeInfo);
if (StringUtils.isNotBlank(entity.getExtParams())) {
IcebergDataNodeDTO dto = IcebergDataNodeDTO.getFromJson(entity.getExtParams());
CommonBeanUtils.copyProperties(dto, icebergDataNodeInfo);
}
-
- LOGGER.debug("success to get iceberg data node from entity");
return icebergDataNodeInfo;
}
@@ -81,8 +78,8 @@ public class IcebergDataNodeOperator extends AbstractDataNodeOperator {
IcebergDataNodeDTO dto = IcebergDataNodeDTO.getFromRequest(icebergDataNodeRequest);
targetEntity.setExtParams(objectMapper.writeValueAsString(dto));
} catch (Exception e) {
- LOGGER.error("failed to set entity for iceberg data node: ", e);
- throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT.getMessage());
+ throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT,
+ String.format("Failed to build extParams for Iceberg node: %s", e.getMessage()));
}
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/mysql/MySQLDataNodeOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/mysql/MySQLDataNodeOperator.java
index 9848b23c3..8e3f2eef8 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/mysql/MySQLDataNodeOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/mysql/MySQLDataNodeOperator.java
@@ -72,8 +72,6 @@ public class MySQLDataNodeOperator extends AbstractDataNodeOperator {
MySQLDataNodeDTO dto = MySQLDataNodeDTO.getFromJson(entity.getExtParams());
CommonBeanUtils.copyProperties(dto, dataNodeInfo);
}
-
- LOGGER.debug("success to get MySQL data node from entity");
return dataNodeInfo;
}
@@ -84,10 +82,9 @@ public class MySQLDataNodeOperator extends AbstractDataNodeOperator {
try {
MySQLDataNodeDTO dto = MySQLDataNodeDTO.getFromRequest(dataNodeRequest);
targetEntity.setExtParams(objectMapper.writeValueAsString(dto));
- LOGGER.debug("success to set entity for MySQL data node");
} catch (Exception e) {
- LOGGER.error("failed to set entity for MySQL data node: ", e);
- throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT.getMessage());
+ throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT,
+ String.format("Failed to build extParams for MySQL node: %s", e.getMessage()));
}
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/starrocks/StarRocksDataNodeOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/starrocks/StarRocksDataNodeOperator.java
index e4249ac08..1d658d555 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/starrocks/StarRocksDataNodeOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/starrocks/StarRocksDataNodeOperator.java
@@ -69,8 +69,6 @@ public class StarRocksDataNodeOperator extends AbstractDataNodeOperator {
StarRocksDataNodeDTO dto = StarRocksDataNodeDTO.getFromJson(entity.getExtParams());
CommonBeanUtils.copyProperties(dto, starRocksDataNodeInfo);
}
-
- LOGGER.debug("success to get starRocks data node from entity");
return starRocksDataNodeInfo;
}
@@ -81,10 +79,9 @@ public class StarRocksDataNodeOperator extends AbstractDataNodeOperator {
try {
StarRocksDataNodeDTO dto = StarRocksDataNodeDTO.getFromRequest(starRocksDataNodeRequest);
targetEntity.setExtParams(objectMapper.writeValueAsString(dto));
- LOGGER.debug("success to set entity for starRocks data node");
} catch (Exception e) {
- LOGGER.error("failed to set entity for starRocks data node: ", e);
- throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT.getMessage());
+ throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT,
+ String.format("Failed to build extParams for StarRocks node: %s", e.getMessage()));
}
}
diff --git a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/DataNodeController.java b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/DataNodeController.java
index 33ddba8c8..977525975 100644
--- a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/DataNodeController.java
+++ b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/DataNodeController.java
@@ -22,7 +22,9 @@ import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.apache.inlong.manager.common.enums.OperationType;
-import org.apache.inlong.manager.common.validation.UpdateValidation;
+import org.apache.inlong.manager.common.validation.SaveValidation;
+import org.apache.inlong.manager.common.validation.UpdateByIdValidation;
+import org.apache.inlong.manager.common.validation.UpdateByKeyValidation;
import org.apache.inlong.manager.pojo.common.PageResult;
import org.apache.inlong.manager.pojo.common.Response;
import org.apache.inlong.manager.pojo.common.UpdateResult;
@@ -60,7 +62,7 @@ public class DataNodeController {
@ApiOperation(value = "Save node")
@OperationLog(operation = OperationType.CREATE)
@RequiresRoles(value = UserRoleCode.ADMIN)
- public Response<Integer> save(@Validated @RequestBody DataNodeRequest request) {
+ public Response<Integer> save(@Validated(SaveValidation.class) @RequestBody DataNodeRequest request) {
String currentUser = LoginUserUtils.getLoginUser().getName();
return Response.success(dataNodeService.save(request, currentUser));
}
@@ -81,7 +83,7 @@ public class DataNodeController {
@PostMapping(value = "/node/update")
@OperationLog(operation = OperationType.UPDATE)
@ApiOperation(value = "Update data node")
- public Response<Boolean> update(@Validated(UpdateValidation.class) @RequestBody DataNodeRequest request) {
+ public Response<Boolean> update(@Validated(UpdateByIdValidation.class) @RequestBody DataNodeRequest request) {
String username = LoginUserUtils.getLoginUser().getName();
return Response.success(dataNodeService.update(request, username));
}
@@ -89,7 +91,8 @@ public class DataNodeController {
@PostMapping(value = "/node/updateByKey")
@OperationLog(operation = OperationType.UPDATE)
@ApiOperation(value = "Update data node by key")
- public Response<UpdateResult> updateByKey(@RequestBody DataNodeRequest request) {
+ public Response<UpdateResult> updateByKey(
+ @Validated(UpdateByKeyValidation.class) @RequestBody DataNodeRequest request) {
String username = LoginUserUtils.getLoginUser().getName();
return Response.success(dataNodeService.updateByKey(request, username));
}
diff --git a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/OpenDataNodeController.java b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/OpenDataNodeController.java
index 11880782c..9fd214166 100644
--- a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/OpenDataNodeController.java
+++ b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/OpenDataNodeController.java
@@ -18,16 +18,15 @@
package org.apache.inlong.manager.web.controller.openapi;
import org.apache.inlong.manager.common.enums.OperationType;
-import org.apache.inlong.manager.common.validation.UpdateValidation;
+import org.apache.inlong.manager.common.validation.SaveValidation;
+import org.apache.inlong.manager.common.validation.UpdateByIdValidation;
import org.apache.inlong.manager.pojo.common.Response;
import org.apache.inlong.manager.pojo.node.DataNodeInfo;
import org.apache.inlong.manager.pojo.node.DataNodePageRequest;
import org.apache.inlong.manager.pojo.node.DataNodeRequest;
-import org.apache.inlong.manager.pojo.user.UserRoleCode;
import org.apache.inlong.manager.service.node.DataNodeService;
import org.apache.inlong.manager.service.operationlog.OperationLog;
import org.apache.inlong.manager.service.user.LoginUserUtils;
-import org.apache.shiro.authz.annotation.RequiresRoles;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.DeleteMapping;
@@ -71,16 +70,14 @@ public class OpenDataNodeController {
@PostMapping(value = "/node/save")
@ApiOperation(value = "Save node")
@OperationLog(operation = OperationType.CREATE)
- @RequiresRoles(value = UserRoleCode.ADMIN)
- public Response<Integer> save(@Validated @RequestBody DataNodeRequest request) {
+ public Response<Integer> save(@Validated(SaveValidation.class) @RequestBody DataNodeRequest request) {
return Response.success(dataNodeService.save(request, LoginUserUtils.getLoginUser()));
}
@PostMapping(value = "/node/update")
- @OperationLog(operation = OperationType.UPDATE)
@ApiOperation(value = "Update data node")
- @RequiresRoles(value = UserRoleCode.ADMIN)
- public Response<Boolean> update(@Validated(UpdateValidation.class) @RequestBody DataNodeRequest request) {
+ @OperationLog(operation = OperationType.UPDATE)
+ public Response<Boolean> update(@Validated(UpdateByIdValidation.class) @RequestBody DataNodeRequest request) {
return Response.success(dataNodeService.update(request, LoginUserUtils.getLoginUser()));
}
@@ -88,7 +85,6 @@ public class OpenDataNodeController {
@ApiOperation(value = "Delete data node by id")
@OperationLog(operation = OperationType.DELETE)
@ApiImplicitParam(name = "id", value = "Data node ID", dataTypeClass = Integer.class, required = true)
- @RequiresRoles(value = UserRoleCode.ADMIN)
public Response<Boolean> delete(@PathVariable Integer id) {
return Response.success(dataNodeService.delete(id, LoginUserUtils.getLoginUser()));
}
diff --git a/inlong-manager/manager-web/src/test/java/org/apache/inlong/manager/web/controller/DataNodeControllerTest.java b/inlong-manager/manager-web/src/test/java/org/apache/inlong/manager/web/controller/DataNodeControllerTest.java
index a63f9ae2a..1214079b6 100644
--- a/inlong-manager/manager-web/src/test/java/org/apache/inlong/manager/web/controller/DataNodeControllerTest.java
+++ b/inlong-manager/manager-web/src/test/java/org/apache/inlong/manager/web/controller/DataNodeControllerTest.java
@@ -177,7 +177,6 @@ class DataNodeControllerTest extends WebBaseTest {
Response<Boolean> response = getResBody(mvcResult, Boolean.class);
Assertions.assertFalse(response.isSuccess());
- Assertions.assertEquals("id: must not be null\n", response.getErrMsg());
}
}