[GitHub] [inlong] leezng opened a new pull request, #6483: [INLONG-6482][Dashboard] Sink management distinguishes between save-only and save-and-submit processes
leezng opened a new pull request, #6483: URL: https://github.com/apache/inlong/pull/6483 - Fixes #6482 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] thesumery opened a new pull request, #6486: [INLONG-6484][Sort] Fix schema update Circular dependency error in multiple sink iceberg scenes
thesumery opened a new pull request, #6486: URL: https://github.com/apache/inlong/pull/6486 [INLONG-6484][Sort] Bugfix: fix schema update Circular dependency error in multiple sink iceberg scenes ### Prepare a Pull Request *(Change the title refer to the following example)* - [INLONG-6484][Sort] Fix schema update Circular dependency error in multiple sink iceberg scenes - Fixes #6484 ### Motivation *Fix schema update Circular dependency error in multiple sink iceberg scenes* ### Modifications *Fix schema update Circular dependency error in multiple sink iceberg scenes* -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] dockerzhang merged pull request #6476: [INLONG-6475][Docker] Add a base flink environment for docker-compose
dockerzhang merged PR #6476: URL: https://github.com/apache/inlong/pull/6476 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[inlong] branch master updated (c8864f8f4 -> 55349721f)
This is an automated email from the ASF dual-hosted git repository. dockerzhang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/inlong.git from c8864f8f4 [INLONG-6379][Sort] Complement iceberg multiple sink metric data compute (#6472) add 55349721f [INLONG-6475][Docker] Add a base flink environment for docker-compose (#6476) No new revisions were added by this update. Summary of changes: docker/docker-compose/docker-compose.yml | 25 - 1 file changed, 24 insertions(+), 1 deletion(-)
[GitHub] [inlong] gosonzhang opened a new pull request, #6488: [INLONG-6406][DataProxy] Should support creating sink dynamically after started (addendum)
gosonzhang opened a new pull request, #6488: URL: https://github.com/apache/inlong/pull/6488 - Fixes #6406 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] healchow merged pull request #6480: [INLONG-6477][Manager] Add consume API in the manager client
healchow merged PR #6480: URL: https://github.com/apache/inlong/pull/6480 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[inlong] branch master updated (55349721f -> 7e391ea96)
This is an automated email from the ASF dual-hosted git repository.
healchow pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
from 55349721f [INLONG-6475][Docker] Add a base flink environment for
docker-compose (#6476)
add 7e391ea96 [INLONG-6477][Manager] Add consume API in the manager client
(#6480)
No new revisions were added by this update.
Summary of changes:
.../inlong/manager/client/api/InlongConsume.java} | 60 ++--
.../manager/client/api/impl/InlongConsumeImpl.java | 89
.../client/api/inner/client/ClientFactory.java | 2 +
.../api/inner/client/InlongConsumeClient.java | 147 +++
.../{DataNodeApi.java => InlongConsumeApi.java}| 39 ++---
.../client/api/inner/ClientFactoryTest.java| 3 +
.../client/api/inner/InlongConsumeClientTest.java | 160 +
.../web/controller/InlongConsumeController.java| 5 +-
8 files changed, 439 insertions(+), 66 deletions(-)
copy
inlong-manager/{manager-service/src/main/java/org/apache/inlong/manager/service/consume/InlongConsumeService.java
=>
manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongConsume.java}
(59%)
create mode 100644
inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongConsumeImpl.java
create mode 100644
inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/InlongConsumeClient.java
copy
inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/{DataNodeApi.java
=> InlongConsumeApi.java} (53%)
create mode 100644
inlong-manager/manager-client/src/test/java/org/apache/inlong/manager/client/api/inner/InlongConsumeClientTest.java
[GitHub] [inlong] dockerzhang merged pull request #6478: [INLONG-6470][Dashboard] Supports management of PostgreSQL source
dockerzhang merged PR #6478: URL: https://github.com/apache/inlong/pull/6478 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[inlong] branch master updated (7e391ea96 -> a5aa351d4)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
from 7e391ea96 [INLONG-6477][Manager] Add consume API in the manager client
(#6480)
add a5aa351d4 [INLONG-6470][Dashboard] Supports management of PostgreSQL
source (#6478)
No new revisions were added by this update.
Summary of changes:
inlong-dashboard/src/locales/cn.json | 11 ++-
inlong-dashboard/src/locales/en.json | 9 ++
.../sources/defaults/{Oracle.ts => PostgreSQL.ts} | 95 ++
.../src/metas/sources/defaults/index.ts| 5 ++
4 files changed, 68 insertions(+), 52 deletions(-)
copy inlong-dashboard/src/metas/sources/defaults/{Oracle.ts => PostgreSQL.ts}
(71%)
[GitHub] [inlong-website] EMsnap merged pull request #586: [INLONG-585][Doc] Add document for multiple sink of Doris
EMsnap merged PR #586: URL: https://github.com/apache/inlong-website/pull/586 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[inlong-website] branch master updated: [INLONG-585][Doc] Add document for multiple sink of Doris (#586)
This is an automated email from the ASF dual-hosted git repository.
zirui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong-website.git
The following commit(s) were added to refs/heads/master by this push:
new 05d012e20a [INLONG-585][Doc] Add document for multiple sink of Doris
(#586)
05d012e20a is described below
commit 05d012e20a3a8c6e392160115ed4c4aaf3f1ab50
Author: kuansix <490305...@qq.com>
AuthorDate: Wed Nov 9 17:39:22 2022 +0800
[INLONG-585][Doc] Add document for multiple sink of Doris (#586)
fix error in kafka doc
Co-authored-by: EMsnap
Co-authored-by: EMsnap
---
docs/data_node/load_node/doris.md | 163 +++--
docs/data_node/load_node/kafka.md | 6 +-
.../current/data_node/load_node/doris.md | 161 ++--
.../current/data_node/load_node/kafka.md | 6 +-
4 files changed, 302 insertions(+), 34 deletions(-)
diff --git a/docs/data_node/load_node/doris.md
b/docs/data_node/load_node/doris.md
index 2facf8e0dd..8dd6a15770 100644
--- a/docs/data_node/load_node/doris.md
+++ b/docs/data_node/load_node/doris.md
@@ -6,9 +6,11 @@ sidebar_position: 16
import {siteVariables} from '../../version';
## Overview
-
-The `Doris Load` node supports writing data to the Doris database. This
document describes how to set up a
-Doris Load node to run SQL queries against the Doris database.
+ - `Doris Load` node supports writing data to the Doris database.
+ - Two modes are supported for sink to Doris:
+Single-sink for specify fixed database name and table name to sink.
+Multi-sink for custom database name and table name according to src format,
which suitable for scenarios such as multi-table writing or whole database
synchronization.
+ - This document describes how to set up a Doris Load node to sink to Doris.
## Supported Version
@@ -33,11 +35,11 @@ such as Maven or SBT is provided below.
## Prepare
-### Create a MySql Extract table
-First create a table `cdc_mysql_source` in the MySql database, the command is
as follows:
+### Create MySql Extract table
+- For Single-sink: Create a table `cdc.cdc_mysql_source` in the MySQL
database. The command is as follows:
```sql
[root@fe001 ~]# mysql -u root -h localhost -P 3306 -p123456
-mysql> use test;
+mysql> use cdc;
Database changed
mysql> CREATE TABLE `cdc_mysql_source` (
`id` int(11) NOT NULL AUTO_INCREMENT,
@@ -59,18 +61,65 @@ mysql> select * from cdc_mysql_source;
| 2 | lisi | 0 |
| 3 | wangwu | 0 |
++--++
-3 rows in set (0.07 sec)
+3 rows in set (0.07 sec)
+```
+- For Multi-sink: Create tables `user_db.user_id_name`、`user_db.user_id_name`
in the MySQL database. The command is as follows:
+```sql
+[root@fe001 ~]# mysql -u root -h localhost -P 3306 -p123456
+mysql> use user_db;
+Database changed
+mysql> CREATE TABLE `user_id_name` (
+ `id` int(11) NOT NULL AUTO_INCREMENT,
+ `name` varchar(64) DEFAULT NULL
+ PRIMARY KEY (`id`)
+ );
+Query OK, 0 rows affected (0.02 sec)
+
+mysql> CREATE TABLE `user_id_score` (
+ `id` int(11) NOT NULL AUTO_INCREMENT,
+ `score` double default 0,
+ PRIMARY KEY (`id`)
+ );
+Query OK, 0 rows affected (0.02 sec)
+
+mysql> insert into user_id_name values(1001, 'lily'),(1002, 'tom'),(1003,
'alan');
+Query OK, 3 rows affected (0.01 sec)
+Records: 3 Duplicates: 0 Warnings: 0
+
+mysql> insert into user_id_score values(1001, 99),(1002, 96),(1003, 98);
+Query OK, 3 rows affected (0.01 sec)
+Records: 3 Duplicates: 0 Warnings: 0
+
+mysql> select * from user_id_name;
++--++
+| id | name |
++--++
+| 1001 | lily |
+| 1002 | tom|
+| 1003 | alan |
+++--+
+3 rows in set (0.07 sec)
+
+mysql> select * from user_id_score;
++--+--+
+| id | name |
++--+--+
+| 1001 | 99 |
+| 1002 | 96 |
+| 1003 | 98 |
++++
+3 rows in set (0.07 sec)
```
-### Create a Doris Load table
-Create a table `cdc_doris_sink` in the Doris database, the command is as
follows:
+### Create Doris Load table
+- For Single-sink: Create a table `cdc.cdc_doris_sink` in the Doris database.
The command is as follows:
```sql
[root@fe001 ~]# mysql -u root -h localhost -P 9030 -p00
-mysql> use test;
+mysql> use cdc;
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A
-
Database changed
+
mysql> CREATE TABLE `cdc_doris_sink` (
`id` int(11) NOT NULL COMMENT "user id",
`name` varchar(50) NOT NULL COMMENT "user name",
@@ -84,11 +133,43 @@ mysql> CREATE TABLE `cdc_doris_sink` (
);
Query OK, 0 rows affected (0.06 sec)
```
+- For Multi-sink: Create tables
`user_db.doris_user_id_name`、`user_db.doris_user_id_score` in the Doris
database. The command is as follows:
+```sql
+[root@fe001 ~]# mysql -u root -h localhost -P 9030 -p00
[GitHub] [inlong-website] EMsnap closed issue #585: [Feature][Doc] Add document for multiple sink of Doris
EMsnap closed issue #585: [Feature][Doc] Add document for multiple sink of Doris URL: https://github.com/apache/inlong-website/issues/585 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[inlong] branch master updated (a5aa351d4 -> afcea59b5)
This is an automated email from the ASF dual-hosted git repository. leezng pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/inlong.git from a5aa351d4 [INLONG-6470][Dashboard] Supports management of PostgreSQL source (#6478) add afcea59b5 [INLONG-6482][Dashboard] Sink management distinguishes between save-only and save-and-submit processes (#6483) No new revisions were added by this update. Summary of changes: inlong-dashboard/src/locales/cn.json | 3 +++ inlong-dashboard/src/locales/en.json | 3 +++ .../src/metas/groups/defaults/index.ts | 2 +- .../pages/GroupDetail/DataStorage/DetailModal.tsx | 24 +++--- .../src/pages/GroupDetail/DataStorage/index.tsx| 1 + 5 files changed, 29 insertions(+), 4 deletions(-)
[GitHub] [inlong] leezng merged pull request #6483: [INLONG-6482][Dashboard] Sink management distinguishes between save-only and save-and-submit processes
leezng merged PR #6483: URL: https://github.com/apache/inlong/pull/6483 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] dockerzhang merged pull request #6464: [INLONG-6463][Manager] Support create subscription and topic of multiple pulsar cluster
dockerzhang merged PR #6464: URL: https://github.com/apache/inlong/pull/6464 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[inlong] branch master updated: [INLONG-6463][Manager] Support create subscription and topic of multiple pulsar cluster (#6464)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new b951f5f0a [INLONG-6463][Manager] Support create subscription and topic
of multiple pulsar cluster (#6464)
b951f5f0a is described below
commit b951f5f0a8d72f7e6761022e4eba756f2582a35d
Author: vernedeng
AuthorDate: Wed Nov 9 17:48:51 2022 +0800
[INLONG-6463][Manager] Support create subscription and topic of multiple
pulsar cluster (#6464)
---
.../queue/pulsar/PulsarResourceOperator.java | 163 -
1 file changed, 95 insertions(+), 68 deletions(-)
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarResourceOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarResourceOperator.java
index 5b698b856..23c13b740 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarResourceOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarResourceOperator.java
@@ -27,7 +27,6 @@ import org.apache.inlong.manager.common.enums.ClusterType;
import org.apache.inlong.manager.common.enums.GroupStatus;
import org.apache.inlong.manager.common.exceptions.WorkflowListenerException;
import org.apache.inlong.manager.common.util.Preconditions;
-import org.apache.inlong.manager.pojo.cluster.ClusterInfo;
import org.apache.inlong.manager.pojo.cluster.pulsar.PulsarClusterInfo;
import org.apache.inlong.manager.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.pojo.group.pulsar.InlongPulsarInfo;
@@ -45,6 +44,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
+import java.util.stream.Collectors;
/**
* Operator for create Pulsar Tenant, Namespace, Topic and Subscription
@@ -84,42 +84,51 @@ public class PulsarResourceOperator implements
QueueResourceOperator {
// get pulsar cluster via the inlong cluster tag from the inlong group
String clusterTag = groupInfo.getInlongClusterTag();
-PulsarClusterInfo pulsarCluster = (PulsarClusterInfo)
clusterService.getOne(clusterTag, null,
-ClusterType.PULSAR);
-try (PulsarAdmin pulsarAdmin =
PulsarUtils.getPulsarAdmin(pulsarCluster)) {
-// create pulsar tenant and namespace
-String tenant = pulsarCluster.getTenant();
-if (StringUtils.isEmpty(tenant)) {
-tenant = InlongConstants.DEFAULT_PULSAR_TENANT;
-}
-String namespace = groupInfo.getMqResource();
-InlongPulsarInfo pulsarInfo = (InlongPulsarInfo) groupInfo;
-// if the group was not successful, need create tenant and
namespace
-if (!Objects.equal(GroupStatus.CONFIG_SUCCESSFUL.getCode(),
groupInfo.getStatus())) {
-pulsarOperator.createTenant(pulsarAdmin, tenant);
-log.info("success to create pulsar tenant for groupId={},
tenant={}", groupId, tenant);
-pulsarOperator.createNamespace(pulsarAdmin, pulsarInfo,
tenant, namespace);
-log.info("success to create pulsar namespace for groupId={},
namespace={}", groupId, namespace);
-}
+List pulsarClusters =
+clusterService.listByTagAndType(clusterTag,
ClusterType.PULSAR).stream()
+.map(clusterInfo -> (PulsarClusterInfo) clusterInfo)
+.collect(Collectors.toList());
+for (PulsarClusterInfo pulsarCluster : pulsarClusters) {
+try (PulsarAdmin pulsarAdmin =
PulsarUtils.getPulsarAdmin(pulsarCluster)) {
+String clusterName = pulsarCluster.getName();
+// create pulsar tenant and namespace
+String tenant = pulsarCluster.getTenant();
+if (StringUtils.isEmpty(tenant)) {
+tenant = InlongConstants.DEFAULT_PULSAR_TENANT;
+}
+String namespace = groupInfo.getMqResource();
+InlongPulsarInfo pulsarInfo = (InlongPulsarInfo) groupInfo;
+// if the group was not successful, need create tenant and
namespace
+if (!Objects.equal(GroupStatus.CONFIG_SUCCESSFUL.getCode(),
groupInfo.getStatus())) {
+pulsarOperator.createTenant(pulsarAdmin, tenant);
+log.info("success to create pulsar tenant for groupId={},
tenant={}, cluster={}",
+groupId, tenant, clusterName);
+pulsarOperator.createNamespace(pulsarAdmin, pulsarInfo,
tenant, namespace);
+log.info("success to create pulsar na
[GitHub] [inlong] EMsnap merged pull request #6203: [INLONG-6176][Agent] Support collect data from Oracle
EMsnap merged PR #6203: URL: https://github.com/apache/inlong/pull/6203 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[inlong] branch master updated: [INLONG-6176][Agent] Support collect data from Oracle (#6203)
This is an automated email from the ASF dual-hosted git repository.
zirui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new cc3da3797 [INLONG-6176][Agent] Support collect data from Oracle (#6203)
cc3da3797 is described below
commit cc3da379709a767e277a756500371aa2333946bd
Author: haibo.duan
AuthorDate: Wed Nov 9 18:35:16 2022 +0800
[INLONG-6176][Agent] Support collect data from Oracle (#6203)
---
.../inlong/agent/constant/OracleConstants.java | 50 ++
.../apache/inlong/agent/pojo/JobProfileDto.java| 53 +-
.../org/apache/inlong/agent/pojo/OracleJob.java| 75 +++
inlong-agent/agent-plugins/pom.xml | 5 +
.../inlong/agent/plugin/sources/OracleSource.java | 48 ++
.../plugin/sources/reader/AbstractReader.java | 35 ++
.../agent/plugin/sources/reader/BinlogReader.java | 32 +-
.../agent/plugin/sources/reader/MongoDBReader.java | 32 +-
.../{SQLServerReader.java => OracleReader.java}| 634 ++---
.../plugin/sources/reader/PostgreSQLReader.java| 32 +-
.../plugin/sources/reader/SQLServerReader.java | 34 +-
.../sources/snapshot/OracleSnapshotBase.java | 52 ++
.../agent/plugin/sources/TestOracleConnect.java| 62 ++
.../agent/plugin/sources/TestOracleReader.java | 234
.../agent/plugin/sources/TestOracleSource.java | 90 +++
pom.xml| 7 +
16 files changed, 1037 insertions(+), 438 deletions(-)
diff --git
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/OracleConstants.java
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/OracleConstants.java
new file mode 100644
index 0..6ba84a7c3
--- /dev/null
+++
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/OracleConstants.java
@@ -0,0 +1,50 @@
+/*
+ * 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.agent.constant;
+
+public class OracleConstants {
+
+/**
+ * The snapshot includes the structure and data of the captured tables.
+ * Specify this value to populate topics with a complete representation of
the data from the captured tables.
+ */
+public static final String INITIAL = "initial";
+
+/**
+ * The snapshot includes the structure and data of the captured tables.
+ * The connector performs an initial snapshot and then stops, without
processing any subsequent changes.
+ */
+public static final String INITIAL_ONLY = "initial_only";
+
+/**
+ * The snapshot includes only the structure of captured tables.
+ * Specify this value if you want the connector to capture data only for
changes that occur after the snapshot.
+ */
+public static final String SCHEMA_ONLY = "schema_only";
+
+/**
+ * This is a recovery setting for a connector that has already been
capturing changes.
+ * When you restart the connector, this setting enables recovery of a
corrupted or lost database history topic.
+ * You might set it periodically to "clean up" a database history topic
that has been growing unexpectedly.
+ * Database history topics require infinite retention. Note this mode is
only safe to be used when it is guaranteed
+ * that no schema changes happened since the point in time the connector
was shut down before and the point in time
+ * the snapshot is taken.
+ */
+public static final String SCHEMA_ONLY_RECOVERY = "schema_only_recovery";
+
+}
diff --git
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
index 810a42d2b..f967dd1a8 100644
---
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
+++
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
@@ -54,6 +54,10 @@ public class JobProfileDto {
* mongo source
*/
public static final String MONGO_SOURCE =
"org.apache.inlong.agent.plugin.sources.MongoDBSource";
+/**
+ * oracl
[inlong] branch master updated (cc3da3797 -> 56bd08bc1)
This is an automated email from the ASF dual-hosted git repository. gosonzhang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/inlong.git from cc3da3797 [INLONG-6176][Agent] Support collect data from Oracle (#6203) add 56bd08bc1 [INLONG-6406][DataProxy] Should support creating sink dynamically after started (addendum) (#6488) No new revisions were added by this update. Summary of changes: .../org/apache/inlong/dataproxy/sink/PulsarSink.java | 5 + .../org/apache/inlong/dataproxy/sink/TubeSink.java | 20 +++- .../dataproxy/sink/common/TubeProducerHolder.java| 6 +- 3 files changed, 13 insertions(+), 18 deletions(-)
[GitHub] [inlong] gosonzhang merged pull request #6488: [INLONG-6406][DataProxy] Should support creating sink dynamically after started (addendum)
gosonzhang merged PR #6488: URL: https://github.com/apache/inlong/pull/6488 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] bluewang opened a new pull request, #6490: [INLONG-6481][Dashboard] Supports management of SQLServer source
bluewang opened a new pull request, #6490: URL: https://github.com/apache/inlong/pull/6490 ### Prepare a Pull Request *(Change the title refer to the following example)* - Title Example: [INLONG-XYZ][Component] Title of the pull request *(The following *XYZ* should be replaced by the actual [GitHub Issue](https://github.com/apache/inlong/issues) number)* - Fixes https://github.com/apache/inlong/issues/6481 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
healchow commented on code in PR #6489:
URL: https://github.com/apache/inlong/pull/6489#discussion_r1017794135
##
inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamSourceController.java:
##
@@ -85,4 +87,15 @@ public Response delete(@PathVariable Integer id) {
return Response.success(result);
}
-}
\ No newline at end of file
+@RequestMapping(value = "/source/delete/forceDelete", method =
RequestMethod.DELETE)
Review Comment:
`/source/delete/forceDelete` -> `/source/forceDelete`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
healchow commented on code in PR #6489:
URL: https://github.com/apache/inlong/pull/6489#discussion_r1017796191
##
inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamSourceEntityMapper.java:
##
@@ -105,6 +105,9 @@ List
selectByStatusAndIp(@Param("statusList") List
int updateByPrimaryKeySelective(StreamSourceEntity record);
+int updateByGroupIdAndStreamId(@Param("groupId") String groupId,
@Param("streamId") String streamId,
Review Comment:
`updateByGroupIdAndStreamId` -> `updateByRelatedId`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
healchow commented on code in PR #6489:
URL: https://github.com/apache/inlong/pull/6489#discussion_r1017796484
##
inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamSourceFieldEntityMapper.java:
##
@@ -59,6 +59,13 @@ public interface StreamSourceFieldEntityMapper {
*/
int deleteAll(@Param("sourceId") Integer sourceId);
+/**
+ * Physically delete all stream source fields based on inlong group ids
+ *
+ * @return rows deleted
+ */
+int deleteByGroupIdAndStreamId(@Param("groupId") String groupId,
@Param("streamId") String streamId);
Review Comment:
`deleteByGroupIdAndStreamId` -> `deleteByRelatedId`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
healchow commented on code in PR #6489:
URL: https://github.com/apache/inlong/pull/6489#discussion_r1017797558
##
inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamSourceFieldEntityMapper.java:
##
@@ -59,6 +59,13 @@ public interface StreamSourceFieldEntityMapper {
*/
int deleteAll(@Param("sourceId") Integer sourceId);
+/**
+ * Physically delete all stream source fields based on inlong group ids
Review Comment:
The java doc was incorrect.
`Physically delete all stream source fields based on inlong group id and
inlong stream id`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] fuweng11 commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
fuweng11 commented on code in PR #6489:
URL: https://github.com/apache/inlong/pull/6489#discussion_r1017803398
##
inlong-manager/manager-dao/src/main/resources/mappers/StreamSourceEntityMapper.xml:
##
@@ -271,6 +271,18 @@
+
+update stream_source
+
+is_deleted = id,
+previous_status = status,
+status = #{status, jdbcType=INTEGER},
Review Comment:
It is recommended to modify the `version` to prevent the deleted source from
being updated during concurrent modification.
##
inlong-manager/manager-dao/src/main/resources/mappers/StreamSourceEntityMapper.xml:
##
@@ -271,6 +271,18 @@
+
Review Comment:
It is recommended to modify the method name, indicating that this method is
a delete operation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[inlong] branch branch-1.4 updated (a39c8fd9b -> 287644e7c)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a change to branch branch-1.4
in repository https://gitbox.apache.org/repos/asf/inlong.git
from a39c8fd9b [INLONG-6379][Sort] Complement iceberg multiple sink metric
data compute (#6472)
new ec6d0ca59 [INLONG-6475][Docker] Add a base flink environment for
docker-compose (#6476)
new f75539106 [INLONG-6477][Manager] Add consume API in the manager client
(#6480)
new a5aa18265 [INLONG-6470][Dashboard] Supports management of PostgreSQL
source (#6478)
new 1d298a1b8 [INLONG-6482][Dashboard] Sink management distinguishes
between save-only and save-and-submit processes (#6483)
new d4d737167 [INLONG-6463][Manager] Support create subscription and topic
of multiple pulsar cluster (#6464)
new 16dfc8afc [INLONG-6176][Agent] Support collect data from Oracle (#6203)
new 287644e7c [INLONG-6406][DataProxy] Should support creating sink
dynamically after started (addendum) (#6488)
The 7 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:
docker/docker-compose/docker-compose.yml | 25 +-
.../inlong/agent/constant/OracleConstants.java | 50 ++
.../apache/inlong/agent/pojo/JobProfileDto.java| 53 +-
.../pojo/{SqlServerJob.java => OracleJob.java} | 11 +-
inlong-agent/agent-plugins/pom.xml | 5 +
.../{SQLServerSource.java => OracleSource.java}| 96 ++--
.../plugin/sources/reader/AbstractReader.java | 35 ++
.../agent/plugin/sources/reader/BinlogReader.java | 32 +-
.../agent/plugin/sources/reader/MongoDBReader.java | 32 +-
.../{SQLServerReader.java => OracleReader.java}| 634 ++---
.../plugin/sources/reader/PostgreSQLReader.java| 32 +-
.../plugin/sources/reader/SQLServerReader.java | 34 +-
...erSnapshotBase.java => OracleSnapshotBase.java} | 8 +-
...QLServerConnect.java => TestOracleConnect.java} | 127 ++---
...tSQLServerReader.java => TestOracleReader.java} | 83 ++-
...tSQLServerSource.java => TestOracleSource.java} | 180 +++---
inlong-dashboard/src/locales/cn.json | 14 +-
inlong-dashboard/src/locales/en.json | 12 +
.../src/metas/groups/defaults/index.ts | 2 +-
.../sources/defaults/{Oracle.ts => PostgreSQL.ts} | 95 ++-
.../src/metas/sources/defaults/index.ts| 5 +
.../pages/GroupDetail/DataStorage/DetailModal.tsx | 24 +-
.../src/pages/GroupDetail/DataStorage/index.tsx| 1 +
.../apache/inlong/dataproxy/sink/PulsarSink.java | 5 +-
.../org/apache/inlong/dataproxy/sink/TubeSink.java | 20 +-
.../dataproxy/sink/common/TubeProducerHolder.java | 6 +-
.../inlong/manager/client/api/InlongConsume.java} | 60 +-
.../manager/client/api/impl/InlongConsumeImpl.java | 89 +++
.../client/api/inner/client/ClientFactory.java | 2 +
.../api/inner/client/InlongConsumeClient.java | 147 +
.../{DataNodeApi.java => InlongConsumeApi.java}| 39 +-
.../client/api/inner/ClientFactoryTest.java| 3 +
.../client/api/inner/InlongConsumeClientTest.java | 160 ++
.../queue/pulsar/PulsarResourceOperator.java | 163 +++---
.../web/controller/InlongConsumeController.java| 5 +-
pom.xml| 7 +
36 files changed, 1394 insertions(+), 902 deletions(-)
create mode 100644
inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/OracleConstants.java
copy
inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/{SqlServerJob.java
=> OracleJob.java} (90%)
copy
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/{SQLServerSource.java
=> OracleSource.java} (80%)
copy
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/{SQLServerReader.java
=> OracleReader.java} (72%)
copy
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/snapshot/{SqlServerSnapshotBase.java
=> OracleSnapshotBase.java} (88%)
copy
inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sources/{TestSQLServerConnect.java
=> TestOracleConnect.java} (68%)
copy
inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sources/{TestSQLServerReader.java
=> TestOracleReader.java} (68%)
copy
inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sources/{TestSQLServerSource.java
=> TestOracleSource.java} (90%)
copy inlong-dashboard/src/metas/sources/defaults/{Oracle.ts => PostgreSQL.ts}
(71%)
copy
inlong-manager/{manager-service/src/main/java/org/apache/inlong/manager/service/consume/InlongConsumeService.java
=>
manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongConsume.java}
(59%)
create mode 100644
inlon
[inlong] 07/07: [INLONG-6406][DataProxy] Should support creating sink dynamically after started (addendum) (#6488)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.4
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 287644e7c8db46771621ca735467624a966c0058
Author: Goson Zhang <4675...@qq.com>
AuthorDate: Wed Nov 9 18:40:10 2022 +0800
[INLONG-6406][DataProxy] Should support creating sink dynamically after
started (addendum) (#6488)
---
.../org/apache/inlong/dataproxy/sink/PulsarSink.java | 5 +
.../org/apache/inlong/dataproxy/sink/TubeSink.java | 20 +++-
.../dataproxy/sink/common/TubeProducerHolder.java| 6 +-
3 files changed, 13 insertions(+), 18 deletions(-)
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java
index 6fd938030..0e0330231 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java
@@ -306,7 +306,6 @@ public class PulsarSink extends AbstractSink implements
Configurable, SendMessag
pulsarCluster = configManager.getMqClusterUrl2Token();
if (!ConfigManager.getInstance().isMqClusterReady()) {
-this.canTake = true;
ConfigManager.getInstance().updMqClusterStatus(true);
logger.info("[{}] MQ Cluster service status ready!", getName());
}
@@ -350,9 +349,7 @@ public class PulsarSink extends AbstractSink implements
Configurable, SendMessag
ConfigConstants.DEFAULT_PROXY_CLUSTER_NAME);
this.metricItemSet = new DataProxyMetricItemSet(clusterId,
this.getName());
MetricRegister.register(metricItemSet);
-if (ConfigManager.getInstance().isMqClusterReady()) {
-this.canTake = true;
-}
+this.canTake = true;
logger.info("[{}] Pulsar sink started", getName());
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/TubeSink.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/TubeSink.java
index dee798214..e1dbf90a3 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/TubeSink.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/TubeSink.java
@@ -76,7 +76,6 @@ public class TubeSink extends AbstractSink implements
Configurable {
private static final Logger logger =
LoggerFactory.getLogger(TubeSink.class);
private static final MsgDedupHandler MSG_DEDUP_HANDLER = new
MsgDedupHandler();
private TubeProducerHolder producerHolder = null;
-private volatile boolean canTake = false;
private volatile boolean canSend = false;
private volatile boolean isOverFlow = false;
private ConfigManager configManager;
@@ -212,9 +211,6 @@ public class TubeSink extends AbstractSink implements
Configurable {
// start the cleaner thread
super.start();
this.canSend = true;
-if (ConfigManager.getInstance().isMqClusterReady()) {
-this.canTake = true;
-}
for (int i = 0; i < sinkThreadPool.length; i++) {
sinkThreadPool[i] = new Thread(new TubeSinkTask(),
getName() + "_tube_sink_sender-" + i);
@@ -229,7 +225,6 @@ public class TubeSink extends AbstractSink implements
Configurable {
logger.info("Duplicated call, " + getName() + " has stopped!");
return;
}
-this.canTake = false;
// waiting inflight message processed
int waitCount = 0;
while (takenMsgCnt.get() > 0 && waitCount++ < 10) {
@@ -269,7 +264,7 @@ public class TubeSink extends AbstractSink implements
Configurable {
@Override
public Status process() throws EventDeliveryException {
-if (!this.canTake) {
+if (!this.started.get()) {
return Status.BACKOFF;
}
Status status = Status.READY;
@@ -319,6 +314,10 @@ public class TubeSink extends AbstractSink implements
Configurable {
logger.info("sink task {} started.",
Thread.currentThread().getName());
while (canSend) {
try {
+if (!started.get() && cachedMsgCnt.get() <= 0) {
+logger.info("Found started is false and taken message
count is zero, braek!");
+break;
+}
if (isOverFlow) {
isOverFlow = false;
Thread.sleep(30);
@@ -327,10 +326,6 @@ public class TubeSink extends AbstractSink implements
Configurable {
if (resendQueue.isEmpty()) {
event = eventQueue.poll(2000, TimeUnit.MILLISECONDS);
[inlong] 04/07: [INLONG-6482][Dashboard] Sink management distinguishes between save-only and save-and-submit processes (#6483)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.4
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 1d298a1b8a26c6e1f707bdb84bc83198edf92f79
Author: Daniel
AuthorDate: Wed Nov 9 17:46:10 2022 +0800
[INLONG-6482][Dashboard] Sink management distinguishes between save-only
and save-and-submit processes (#6483)
---
inlong-dashboard/src/locales/cn.json | 3 +++
inlong-dashboard/src/locales/en.json | 3 +++
.../src/metas/groups/defaults/index.ts | 2 +-
.../pages/GroupDetail/DataStorage/DetailModal.tsx | 24 +++---
.../src/pages/GroupDetail/DataStorage/index.tsx| 1 +
5 files changed, 29 insertions(+), 4 deletions(-)
diff --git a/inlong-dashboard/src/locales/cn.json
b/inlong-dashboard/src/locales/cn.json
index e99371b9c..44ad7e605 100644
--- a/inlong-dashboard/src/locales/cn.json
+++ b/inlong-dashboard/src/locales/cn.json
@@ -384,6 +384,9 @@
"pages.GroupDetail.Sink.Status.Pending": "配置中",
"pages.GroupDetail.Sink.Status.Error": "配置失败",
"pages.GroupDetail.Sink.Status.Success": "配置成功",
+ "pages.GroupDetail.Sink.Cancel": "取消",
+ "pages.GroupDetail.Sink.Save": "仅保存",
+ "pages.GroupDetail.Sink.SaveAndRun": "保存并提交流程",
"pages.GroupDetail.Stream.StreamConfigTitle": "数据流配置",
"pages.GroupDetail.Stream.CreateDataStream": "新建数据流",
"pages.GroupDetail.PageTitle": "详情",
diff --git a/inlong-dashboard/src/locales/en.json
b/inlong-dashboard/src/locales/en.json
index 255f56996..ecdfcfa24 100644
--- a/inlong-dashboard/src/locales/en.json
+++ b/inlong-dashboard/src/locales/en.json
@@ -384,6 +384,9 @@
"pages.GroupDetail.Sink.Status.Pending": "Pending",
"pages.GroupDetail.Sink.Status.Error": "Error",
"pages.GroupDetail.Sink.Status.Success": "Success",
+ "pages.GroupDetail.Sink.Cancel": "Cancel",
+ "pages.GroupDetail.Sink.Save": "Save",
+ "pages.GroupDetail.Sink.SaveAndRun": "Save and submit the process",
"pages.GroupDetail.Stream.StreamConfigTitle": "Data stream configuration",
"pages.GroupDetail.Stream.CreateDataStream": "Create",
"pages.GroupDetail.PageTitle": "Detail",
diff --git a/inlong-dashboard/src/metas/groups/defaults/index.ts
b/inlong-dashboard/src/metas/groups/defaults/index.ts
index e68379f4a..1eacc5e5b 100644
--- a/inlong-dashboard/src/metas/groups/defaults/index.ts
+++ b/inlong-dashboard/src/metas/groups/defaults/index.ts
@@ -37,7 +37,7 @@ export const allDefaultGroups:
MetaExportWithBackendList = [
LoadEntity: () => import('./Pulsar'),
},
{
-label: 'TubeMq',
+label: 'TubeMQ',
value: 'TUBEMQ',
LoadEntity: () => import('./TubeMq'),
},
diff --git a/inlong-dashboard/src/pages/GroupDetail/DataStorage/DetailModal.tsx
b/inlong-dashboard/src/pages/GroupDetail/DataStorage/DetailModal.tsx
index bba88d61b..0e5489545 100644
--- a/inlong-dashboard/src/pages/GroupDetail/DataStorage/DetailModal.tsx
+++ b/inlong-dashboard/src/pages/GroupDetail/DataStorage/DetailModal.tsx
@@ -18,7 +18,7 @@
*/
import React, { useMemo, useState } from 'react';
-import { Skeleton, Modal, message } from 'antd';
+import { Button, Skeleton, Modal, message } from 'antd';
import { ModalProps } from 'antd/es/modal';
import { useRequest, useUpdateEffect } from '@/hooks';
import { useTranslation } from 'react-i18next';
@@ -84,7 +84,7 @@ const Comp: React.FC = ({ inlongGroupId,
id, ...modalProps })
return Entity ? new Entity().renderRow() : [];
}, [Entity]);
- const onOk = async () => {
+ const onOk = async (startProcess = false) => {
const values = await form.validateFields();
const submitData = new Entity()?.stringify(values) || values;
const isUpdate = Boolean(id);
@@ -92,6 +92,9 @@ const Comp: React.FC = ({ inlongGroupId,
id, ...modalProps })
submitData.id = id;
submitData.version = data?.version;
}
+if (startProcess) {
+ submitData.startProcess = true;
+}
await request({
url: isUpdate ? '/sink/update' : '/sink/save',
method: 'POST',
@@ -105,7 +108,22 @@ const Comp: React.FC = ({ inlongGroupId,
id, ...modalProps })
};
return (
-
+
+ {t('pages.GroupDetail.Sink.Cancel')}
+,
+ onOk(false)}>
+ {t('pages.GroupDetail.Sink.Save')}
+,
+ onOk(true)}>
+ {t('pages.GroupDetail.Sink.SaveAndRun')}
+,
+ ]}
+>
{loading ? (
) : (
diff --git a/inlong-dashboard/src/pages/GroupDetail/DataStorage/index.tsx
b/inlong-dashboard/src/pages/GroupDetail/DataStorage/index.tsx
index cdada63ba..3c836da23 100644
--- a/inlong-dashboard/src/pages/GroupDetail/DataStorage/index.tsx
+++ b/inlong-dashboard/src/pages/GroupDetail/DataStorage/index.tsx
@@ -76,6 +76,7 @@ const Comp = ({ inlongGroupId, readonly }: Props, ref) => {
method: 'DELETE',
params: {
sinkType: options.sinkType,
+ startProcess: true,
[inlong] 01/07: [INLONG-6475][Docker] Add a base flink environment for docker-compose (#6476)
This is an automated email from the ASF dual-hosted git repository. dockerzhang pushed a commit to branch branch-1.4 in repository https://gitbox.apache.org/repos/asf/inlong.git commit ec6d0ca59e7da23bbcae7e2fdf2191d41717aaec Author: Charles Zhang AuthorDate: Wed Nov 9 17:31:17 2022 +0800 [INLONG-6475][Docker] Add a base flink environment for docker-compose (#6476) --- docker/docker-compose/docker-compose.yml | 25 - 1 file changed, 24 insertions(+), 1 deletion(-) diff --git a/docker/docker-compose/docker-compose.yml b/docker/docker-compose/docker-compose.yml index 8cce345f2..b3db40fcc 100644 --- a/docker/docker-compose/docker-compose.yml +++ b/docker/docker-compose/docker-compose.yml @@ -55,7 +55,7 @@ services: - USERNAME=root - PASSWORD=inlong - ZK_URL=tubemq-server:2181 - - FLINK_HOST=localhost + - FLINK_HOST=jobmanager - FLINK_PORT=8081 - AUDIT_PROXY_URL=audit:10081 @@ -114,3 +114,26 @@ services: - MQ_TYPE=pulsar ports: - "10081:10081" + + # flink jobmanager + jobmanager: +image: apache/flink:1.13-scala_2.11 +container_name: jobmanager +environment: + - | +FLINK_PROPERTIES= +jobmanager.rpc.address: jobmanager +ports: + - "8081:8081" +command: jobmanager + + # flink taskmanager + taskmanager: +image: apache/flink:1.13-scala_2.11 +container_name: taskmanager +environment: + - | +FLINK_PROPERTIES= +jobmanager.rpc.address: jobmanager +taskmanager.numberOfTaskSlots: 2 +command: taskmanager \ No newline at end of file
[inlong] 05/07: [INLONG-6463][Manager] Support create subscription and topic of multiple pulsar cluster (#6464)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.4
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit d4d737167d2f5df013cb457939f2def6961c04c4
Author: vernedeng
AuthorDate: Wed Nov 9 17:48:51 2022 +0800
[INLONG-6463][Manager] Support create subscription and topic of multiple
pulsar cluster (#6464)
---
.../queue/pulsar/PulsarResourceOperator.java | 163 -
1 file changed, 95 insertions(+), 68 deletions(-)
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarResourceOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarResourceOperator.java
index 5b698b856..23c13b740 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarResourceOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarResourceOperator.java
@@ -27,7 +27,6 @@ import org.apache.inlong.manager.common.enums.ClusterType;
import org.apache.inlong.manager.common.enums.GroupStatus;
import org.apache.inlong.manager.common.exceptions.WorkflowListenerException;
import org.apache.inlong.manager.common.util.Preconditions;
-import org.apache.inlong.manager.pojo.cluster.ClusterInfo;
import org.apache.inlong.manager.pojo.cluster.pulsar.PulsarClusterInfo;
import org.apache.inlong.manager.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.pojo.group.pulsar.InlongPulsarInfo;
@@ -45,6 +44,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
+import java.util.stream.Collectors;
/**
* Operator for create Pulsar Tenant, Namespace, Topic and Subscription
@@ -84,42 +84,51 @@ public class PulsarResourceOperator implements
QueueResourceOperator {
// get pulsar cluster via the inlong cluster tag from the inlong group
String clusterTag = groupInfo.getInlongClusterTag();
-PulsarClusterInfo pulsarCluster = (PulsarClusterInfo)
clusterService.getOne(clusterTag, null,
-ClusterType.PULSAR);
-try (PulsarAdmin pulsarAdmin =
PulsarUtils.getPulsarAdmin(pulsarCluster)) {
-// create pulsar tenant and namespace
-String tenant = pulsarCluster.getTenant();
-if (StringUtils.isEmpty(tenant)) {
-tenant = InlongConstants.DEFAULT_PULSAR_TENANT;
-}
-String namespace = groupInfo.getMqResource();
-InlongPulsarInfo pulsarInfo = (InlongPulsarInfo) groupInfo;
-// if the group was not successful, need create tenant and
namespace
-if (!Objects.equal(GroupStatus.CONFIG_SUCCESSFUL.getCode(),
groupInfo.getStatus())) {
-pulsarOperator.createTenant(pulsarAdmin, tenant);
-log.info("success to create pulsar tenant for groupId={},
tenant={}", groupId, tenant);
-pulsarOperator.createNamespace(pulsarAdmin, pulsarInfo,
tenant, namespace);
-log.info("success to create pulsar namespace for groupId={},
namespace={}", groupId, namespace);
-}
+List pulsarClusters =
+clusterService.listByTagAndType(clusterTag,
ClusterType.PULSAR).stream()
+.map(clusterInfo -> (PulsarClusterInfo) clusterInfo)
+.collect(Collectors.toList());
+for (PulsarClusterInfo pulsarCluster : pulsarClusters) {
+try (PulsarAdmin pulsarAdmin =
PulsarUtils.getPulsarAdmin(pulsarCluster)) {
+String clusterName = pulsarCluster.getName();
+// create pulsar tenant and namespace
+String tenant = pulsarCluster.getTenant();
+if (StringUtils.isEmpty(tenant)) {
+tenant = InlongConstants.DEFAULT_PULSAR_TENANT;
+}
+String namespace = groupInfo.getMqResource();
+InlongPulsarInfo pulsarInfo = (InlongPulsarInfo) groupInfo;
+// if the group was not successful, need create tenant and
namespace
+if (!Objects.equal(GroupStatus.CONFIG_SUCCESSFUL.getCode(),
groupInfo.getStatus())) {
+pulsarOperator.createTenant(pulsarAdmin, tenant);
+log.info("success to create pulsar tenant for groupId={},
tenant={}, cluster={}",
+groupId, tenant, clusterName);
+pulsarOperator.createNamespace(pulsarAdmin, pulsarInfo,
tenant, namespace);
+log.info("success to create pulsar namespace for
groupId={}, namespace={}, cluster={}",
+groupId, namespace, clusterName);
+}
-// create pulsar topic - each Inlong Stream corresponds to a
Pu
[inlong] 02/07: [INLONG-6477][Manager] Add consume API in the manager client (#6480)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.4
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit f7553910668072bf0b2719cf028709da2a00eabb
Author: haifxu
AuthorDate: Wed Nov 9 17:37:41 2022 +0800
[INLONG-6477][Manager] Add consume API in the manager client (#6480)
---
.../inlong/manager/client/api/InlongConsume.java | 84 +++
.../manager/client/api/impl/InlongConsumeImpl.java | 89
.../client/api/inner/client/ClientFactory.java | 2 +
.../api/inner/client/InlongConsumeClient.java | 147 +++
.../client/api/service/InlongConsumeApi.java | 59
.../client/api/inner/ClientFactoryTest.java| 3 +
.../client/api/inner/InlongConsumeClientTest.java | 160 +
.../web/controller/InlongConsumeController.java| 5 +-
8 files changed, 546 insertions(+), 3 deletions(-)
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongConsume.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongConsume.java
new file mode 100644
index 0..182d7bae2
--- /dev/null
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongConsume.java
@@ -0,0 +1,84 @@
+/*
+ * 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.client.api;
+
+import org.apache.inlong.manager.pojo.common.PageResult;
+import org.apache.inlong.manager.pojo.consume.InlongConsumeBriefInfo;
+import org.apache.inlong.manager.pojo.consume.InlongConsumeCountInfo;
+import org.apache.inlong.manager.pojo.consume.InlongConsumeInfo;
+import org.apache.inlong.manager.pojo.consume.InlongConsumePageRequest;
+import org.apache.inlong.manager.pojo.consume.InlongConsumeRequest;
+import org.apache.inlong.manager.pojo.workflow.WorkflowResult;
+
+public interface InlongConsume {
+
+/**
+ * Save inlong consume info.
+ *
+ * @param request consume request need to save
+ * @return inlong consume id after saving
+ */
+Integer save(InlongConsumeRequest request);
+
+/**
+ * Get inlong consume info based on ID
+ *
+ * @param id inlong consume id
+ * @return detail of inlong group
+ */
+InlongConsumeInfo get(Integer id);
+
+/**
+ * Query the inlong consume statistics info via the username
+ *
+ * @return inlong consume status statistics
+ */
+InlongConsumeCountInfo countStatusByUser();
+
+/**
+ * Paging query inlong consume info list
+ *
+ * @param request pagination query request
+ * @return inlong consume list
+ */
+PageResult list(InlongConsumePageRequest request);
+
+/**
+ * Update the inlong consume
+ *
+ * @param request inlong consume request that needs to be updated
+ * @return inlong consume id after saving
+ */
+Integer update(InlongConsumeRequest request);
+
+/**
+ * Delete the inlong consume by the id
+ *
+ * @param id inlong consume id that needs to be deleted
+ * @return whether succeed
+ */
+Boolean delete(Integer id);
+
+/**
+ * Start the process for the specified ID.
+ *
+ * @param id inlong consume id
+ * @return workflow result
+ */
+WorkflowResult startProcess(Integer id);
+}
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongConsumeImpl.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongConsumeImpl.java
new file mode 100644
index 0..cc0139d56
--- /dev/null
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongConsumeImpl.java
@@ -0,0 +1,89 @@
+/*
+ * 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 c
[inlong] 03/07: [INLONG-6470][Dashboard] Supports management of PostgreSQL source (#6478)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.4
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit a5aa182651e41a2d568a4b6fe8b55102d765f1f7
Author: Lizhen <88174078+bluew...@users.noreply.github.com>
AuthorDate: Wed Nov 9 17:38:40 2022 +0800
[INLONG-6470][Dashboard] Supports management of PostgreSQL source (#6478)
---
inlong-dashboard/src/locales/cn.json | 11 +-
inlong-dashboard/src/locales/en.json | 9 ++
.../src/metas/sources/defaults/PostgreSQL.ts | 154 +
.../src/metas/sources/defaults/index.ts| 5 +
4 files changed, 178 insertions(+), 1 deletion(-)
diff --git a/inlong-dashboard/src/locales/cn.json
b/inlong-dashboard/src/locales/cn.json
index c5ad9c67a..e99371b9c 100644
--- a/inlong-dashboard/src/locales/cn.json
+++ b/inlong-dashboard/src/locales/cn.json
@@ -44,11 +44,20 @@
"meta.Sources.Oracle.Username": "用户",
"meta.Sources.Oracle.Password": "密码",
"meta.Sources.Oracle.Database": "数据库名",
- "meta.Sources.Oracle.SchemaName": "集合名称",
+ "meta.Sources.Oracle.SchemaName": "架构名称",
"meta.Sources.Oracle.TableName": "表格名称",
"meta.Sources.Oracle.AllMigration": "是否整库迁移",
"meta.Sources.Oracle.ScanStartupMode": "扫描启动模式",
"meta.Sources.Oracle.PrimaryKey": "主键",
+ "meta.Sources.PostgreSQL.Hostname": "服务器主机",
+ "meta.Sources.PostgreSQL.Port": "端口",
+ "meta.Sources.PostgreSQL.Username": "用户",
+ "meta.Sources.PostgreSQL.Password": "密码",
+ "meta.Sources.PostgreSQL.Database": "数据库名",
+ "meta.Sources.PostgreSQL.SchemaName": "架构名称",
+ "meta.Sources.PostgreSQL.TableName": "表格名称",
+ "meta.Sources.PostgreSQL.decodingPluginName": "解码插件名称",
+ "meta.Sources.PostgreSQL.PrimaryKey": "主键",
"meta.Sinks.SinkName": "名称",
"meta.Sinks.SinkNameRule": "以英文字母开头，只能包含英文字母、数字、中划线、下划线",
"meta.Sinks.SinkType": "类型",
diff --git a/inlong-dashboard/src/locales/en.json
b/inlong-dashboard/src/locales/en.json
index dc7f103a1..255f56996 100644
--- a/inlong-dashboard/src/locales/en.json
+++ b/inlong-dashboard/src/locales/en.json
@@ -49,6 +49,15 @@
"meta.Sources.Oracle.AllMigration": "AllMigration",
"meta.Sources.Oracle.ScanStartupMode": "Scan startup mode",
"meta.Sources.Oracle.PrimaryKey": "PrimaryKey",
+ "meta.Sources.PostgreSQL.Hostname": "Hostname",
+ "meta.Sources.PostgreSQL.Port": "Port",
+ "meta.Sources.PostgreSQL.Username": "Username",
+ "meta.Sources.PostgreSQL.Password": "Password",
+ "meta.Sources.PostgreSQL.Database": "Database",
+ "meta.Sources.PostgreSQL.SchemaName": "SchemaName",
+ "meta.Sources.PostgreSQL.TableName": "TableName",
+ "meta.Sources.PostgreSQL.decodingPluginName": "Decoding Plugin Name",
+ "meta.Sources.PostgreSQL.PrimaryKey": "PrimaryKey",
"meta.Sinks.SinkName": "Name",
"meta.Sinks.SinkNameRule": "At the beginning of English letters, only
English letters, numbers, minus, and underscores",
"meta.Sinks.SinkType": "Type",
diff --git a/inlong-dashboard/src/metas/sources/defaults/PostgreSQL.ts
b/inlong-dashboard/src/metas/sources/defaults/PostgreSQL.ts
new file mode 100644
index 0..6bd69cc3f
--- /dev/null
+++ b/inlong-dashboard/src/metas/sources/defaults/PostgreSQL.ts
@@ -0,0 +1,154 @@
+/*
+ * 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.
+ */
+
+import { DataWithBackend } from '@/metas/DataWithBackend';
+import { RenderRow } from '@/metas/RenderRow';
+import { RenderList } from '@/metas/RenderList';
+import { SourceInfo } from '../common/SourceInfo';
+
+const { I18n } = DataWithBackend;
+const { FieldDecorator } = RenderRow;
+const { ColumnDecorator } = RenderList;
+
+export default class PostgreSQLSource
+ extends SourceInfo
+ implements DataWithBackend, RenderRow, RenderList
+{
+ @FieldDecorator({
+type: 'input',
+rules: [{ required: true }],
+props: values => ({
+ disabled: values?.status === 101,
+}),
+ })
+ @ColumnDecorator()
+ @I18n('meta.Sources.PostgreSQL.Hostname')
+ hostname: string;
+
+ @FieldDecorator({
+type: 'inputnumber',
+rules: [{ required: true }],
+initialValue: 5432,
+props: values => ({
+ min: 1,
+ max: 65535,
+ disabled: values?.status === 101,
+}),
[inlong] 06/07: [INLONG-6176][Agent] Support collect data from Oracle (#6203)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.4
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 16dfc8afc748c4a0174a26984d67cd613f83def2
Author: haibo.duan
AuthorDate: Wed Nov 9 18:35:16 2022 +0800
[INLONG-6176][Agent] Support collect data from Oracle (#6203)
---
.../inlong/agent/constant/OracleConstants.java | 50 ++
.../apache/inlong/agent/pojo/JobProfileDto.java| 53 +-
.../org/apache/inlong/agent/pojo/OracleJob.java| 75 +++
inlong-agent/agent-plugins/pom.xml | 5 +
.../inlong/agent/plugin/sources/OracleSource.java | 48 ++
.../plugin/sources/reader/AbstractReader.java | 35 ++
.../agent/plugin/sources/reader/BinlogReader.java | 32 +-
.../agent/plugin/sources/reader/MongoDBReader.java | 32 +-
.../{SQLServerReader.java => OracleReader.java}| 634 ++---
.../plugin/sources/reader/PostgreSQLReader.java| 32 +-
.../plugin/sources/reader/SQLServerReader.java | 34 +-
.../sources/snapshot/OracleSnapshotBase.java | 52 ++
.../agent/plugin/sources/TestOracleConnect.java| 62 ++
.../agent/plugin/sources/TestOracleReader.java | 234
.../agent/plugin/sources/TestOracleSource.java | 90 +++
pom.xml| 7 +
16 files changed, 1037 insertions(+), 438 deletions(-)
diff --git
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/OracleConstants.java
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/OracleConstants.java
new file mode 100644
index 0..6ba84a7c3
--- /dev/null
+++
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/OracleConstants.java
@@ -0,0 +1,50 @@
+/*
+ * 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.agent.constant;
+
+public class OracleConstants {
+
+/**
+ * The snapshot includes the structure and data of the captured tables.
+ * Specify this value to populate topics with a complete representation of
the data from the captured tables.
+ */
+public static final String INITIAL = "initial";
+
+/**
+ * The snapshot includes the structure and data of the captured tables.
+ * The connector performs an initial snapshot and then stops, without
processing any subsequent changes.
+ */
+public static final String INITIAL_ONLY = "initial_only";
+
+/**
+ * The snapshot includes only the structure of captured tables.
+ * Specify this value if you want the connector to capture data only for
changes that occur after the snapshot.
+ */
+public static final String SCHEMA_ONLY = "schema_only";
+
+/**
+ * This is a recovery setting for a connector that has already been
capturing changes.
+ * When you restart the connector, this setting enables recovery of a
corrupted or lost database history topic.
+ * You might set it periodically to "clean up" a database history topic
that has been growing unexpectedly.
+ * Database history topics require infinite retention. Note this mode is
only safe to be used when it is guaranteed
+ * that no schema changes happened since the point in time the connector
was shut down before and the point in time
+ * the snapshot is taken.
+ */
+public static final String SCHEMA_ONLY_RECOVERY = "schema_only_recovery";
+
+}
diff --git
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
index 810a42d2b..f967dd1a8 100644
---
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
+++
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
@@ -54,6 +54,10 @@ public class JobProfileDto {
* mongo source
*/
public static final String MONGO_SOURCE =
"org.apache.inlong.agent.plugin.sources.MongoDBSource";
+/**
+ * oracle source
+ */
+public static final String ORACLE_SOURCE =
"org.apache.inlong.agent.plugin.sources.OracleSource";
/**
* mqtt source
*/
@@ -230,10 +
[GitHub] [inlong] haifxu commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
haifxu commented on code in PR #6489:
URL: https://github.com/apache/inlong/pull/6489#discussion_r1017812972
##
inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/StreamSourceController.java:
##
@@ -85,4 +87,15 @@ public Response delete(@PathVariable Integer id) {
return Response.success(result);
}
-}
\ No newline at end of file
+@RequestMapping(value = "/source/delete/forceDelete", method =
RequestMethod.DELETE)
Review Comment:
Done.
##
inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamSourceEntityMapper.java:
##
@@ -105,6 +105,9 @@ List
selectByStatusAndIp(@Param("statusList") List
int updateByPrimaryKeySelective(StreamSourceEntity record);
+int updateByGroupIdAndStreamId(@Param("groupId") String groupId,
@Param("streamId") String streamId,
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] haifxu commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
haifxu commented on code in PR #6489:
URL: https://github.com/apache/inlong/pull/6489#discussion_r1017813225
##
inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamSourceFieldEntityMapper.java:
##
@@ -59,6 +59,13 @@ public interface StreamSourceFieldEntityMapper {
*/
int deleteAll(@Param("sourceId") Integer sourceId);
+/**
+ * Physically delete all stream source fields based on inlong group ids
+ *
+ * @return rows deleted
+ */
+int deleteByGroupIdAndStreamId(@Param("groupId") String groupId,
@Param("streamId") String streamId);
Review Comment:
Done.
##
inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamSourceFieldEntityMapper.java:
##
@@ -59,6 +59,13 @@ public interface StreamSourceFieldEntityMapper {
*/
int deleteAll(@Param("sourceId") Integer sourceId);
+/**
+ * Physically delete all stream source fields based on inlong group ids
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] haifxu commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
haifxu commented on code in PR #6489:
URL: https://github.com/apache/inlong/pull/6489#discussion_r1017813393
##
inlong-manager/manager-dao/src/main/resources/mappers/StreamSourceEntityMapper.xml:
##
@@ -271,6 +271,18 @@
+
+update stream_source
+
+is_deleted = id,
+previous_status = status,
+status = #{status, jdbcType=INTEGER},
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] haifxu commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
haifxu commented on code in PR #6489: URL: https://github.com/apache/inlong/pull/6489#discussion_r1017814920 ## inlong-manager/manager-dao/src/main/resources/mappers/StreamSourceEntityMapper.xml: ## @@ -271,6 +271,18 @@ + Review Comment: This is a logical delete, maybe `update` would be better. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
healchow commented on code in PR #6489:
URL: https://github.com/apache/inlong/pull/6489#discussion_r1017823900
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java:
##
@@ -267,6 +267,21 @@ public Boolean delete(Integer id, String operator) {
return true;
}
+@Override
+public Boolean forceDelete(String groupId, String streamId, String
operator) {
+LOGGER.info("begin to force delete source for groupId={} and
streamId={} by user={}",
+groupId, streamId, operator);
+Preconditions.checkNotNull(groupId,
ErrorCodeEnum.GROUP_ID_IS_EMPTY.getMessage());
+Preconditions.checkNotNull(streamId,
ErrorCodeEnum.STREAM_ID_IS_EMPTY.getMessage());
+
+sourceMapper.updateByRelatedId(groupId, streamId,
SourceStatus.SOURCE_DISABLE.getCode());
Review Comment:
Suggest printing the affected rows into the log.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] luchunliang opened a new pull request, #6492: [INLONG-6491][Manager] Support getting backup info in getAllConfig
luchunliang opened a new pull request, #6492: URL: https://github.com/apache/inlong/pull/6492 - Title : [INLONG-6491][Manager] Support getting backup info in getAllConfig - Fixes #6491 ### Motivation *Explain here the context, and why you're making that change. What is the problem you're trying to solve?* ### Modifications *Describe the modifications you've done.* ### Verifying this change *(Please pick either of the following options)* - [ ] This change is a trivial rework/code cleanup without any test coverage. - [ ] This change is already covered by existing tests, such as: *(please describe tests)* - [ ] This change added tests and can be verified as follows: *(example:)* - *Added integration tests for end-to-end deployment with large payloads (10MB)* - *Extended integration test for recovery after broker failure* ### Documentation - Does this pull request introduce a new feature? (yes / no) - If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented) - If a feature is not applicable for documentation, explain why? - If a feature is not documented yet in this PR, please create a follow-up issue for adding the documentation -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] thesumery commented on pull request #6486: [INLONG-6484][Sort] Fix schema update Circular dependency error in multiple sink iceberg scenes
thesumery commented on PR #6486: URL: https://github.com/apache/inlong/pull/6486#issuecomment-1309660708 add some unit test for iceberg multiple sink scences: #6485 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] dockerzhang merged pull request #6492: [INLONG-6491][Manager] Support getting backup info in getAllConfig
dockerzhang merged PR #6492: URL: https://github.com/apache/inlong/pull/6492 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[inlong] branch master updated (56bd08bc1 -> 43d4e2aec)
This is an automated email from the ASF dual-hosted git repository. dockerzhang pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/inlong.git from 56bd08bc1 [INLONG-6406][DataProxy] Should support creating sink dynamically after started (addendum) (#6488) add 43d4e2aec [INLONG-6491][Manager] Support getting backup info in getAllConfig (#6492) No new revisions were added by this update. Summary of changes: .../repository/DataProxyConfigRepository.java | 340 +++-- 1 file changed, 171 insertions(+), 169 deletions(-)
[GitHub] [inlong] fuweng11 commented on a diff in pull request #6046: [INLONG-6044][Manager] Distinguish between group and stream configuration processes
fuweng11 commented on code in PR #6046:
URL: https://github.com/apache/inlong/pull/6046#discussion_r1018583409
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/group/DeleteGroupWorkflowDefinition.java:
##
@@ -85,22 +77,12 @@ public WorkflowProcess defineProcess() {
deleteMQTask.setListenerFactory(groupTaskListenerFactory);
process.addTask(deleteMQTask);
-// Delete Sort
Review Comment:
The current sort configuration belongs to the group. The sort configuration
will not be performed until the stream configuration under the current group is
successful.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] vernedeng commented on a diff in pull request #6046: [INLONG-6044][Manager] Distinguish between group and stream configuration processes
vernedeng commented on code in PR #6046:
URL: https://github.com/apache/inlong/pull/6046#discussion_r1018587699
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/kafka/KafkaResourceOperators.java:
##
@@ -67,49 +62,12 @@ public boolean accept(String mqType) {
@Override
public void createQueueForGroup(@NotNull InlongGroupInfo groupInfo,
@NotBlank String operator) {
-String groupId = groupInfo.getInlongGroupId();
-log.info("begin to create kafka resource for groupId={}", groupId);
-
-InlongKafkaInfo inlongKafkaInfo = (InlongKafkaInfo) groupInfo;
-try {
-// 1. create kafka Topic - each Inlong Stream corresponds to a
Kafka Topic
-List streamInfoList =
streamService.getTopicList(groupId);
-if (streamInfoList == null || streamInfoList.isEmpty()) {
-log.warn("skip to create kafka topic and subscription as no
streams for groupId={}", groupId);
-return;
-}
-for (InlongStreamBriefInfo streamInfo : streamInfoList) {
-this.createKafkaTopic(inlongKafkaInfo,
streamInfo.getMqResource());
-}
-} catch (Exception e) {
-String msg = String.format("failed to create kafka resource for
groupId=%s", groupId);
-log.error(msg, e);
-throw new WorkflowListenerException(msg + ": " + e.getMessage());
-}
-log.info("success to create kafka resource for groupId={},
cluster={}", groupId, inlongKafkaInfo);
+log.info("skip to create kafka topic for groupId={}",
groupInfo.getInlongGroupId());
}
@Override
public void deleteQueueForGroup(InlongGroupInfo groupInfo, String
operator) {
-Preconditions.checkNotNull(groupInfo, "inlong group info cannot be
null");
-
-String groupId = groupInfo.getInlongGroupId();
-log.info("begin to delete kafka resource for groupId={}", groupId);
-ClusterInfo clusterInfo =
clusterService.getOne(groupInfo.getInlongClusterTag(), null, ClusterType.KAFKA);
-try {
-List streamInfoList =
streamService.getTopicList(groupId);
-if (streamInfoList == null || streamInfoList.isEmpty()) {
-log.warn("skip to create kafka topic and subscription as no
streams for groupId={}", groupId);
-return;
-}
-for (InlongStreamBriefInfo streamInfo : streamInfoList) {
-this.deleteKafkaTopic(groupInfo,
streamInfo.getInlongStreamId());
-}
-} catch (Exception e) {
-log.error("failed to delete kafka resource for groupId=" +
groupId, e);
-throw new WorkflowListenerException("failed to delete kafka
resource: " + e.getMessage());
-}
-log.info("success to delete kafka resource for groupId={},
cluster={}", groupId, clusterInfo);
+log.info("skip to delete kafka topic for groupId={}",
groupInfo.getInlongGroupId());
Review Comment:
why skip delete kafka topic directly?
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/sort/SortConfigListener.java:
##
@@ -81,6 +82,10 @@ public ListenerResult listen(WorkflowContext context) throws
WorkflowListenerExc
}
InlongGroupInfo groupInfo = form.getGroupInfo();
List streamInfos = form.getStreamInfos();
+if (CollectionUtils.isEmpty(streamInfos)) {
+LOGGER.warn("not build sort config for groupId={}, as the stream
is empty", groupId);
Review Comment:
```suggestion
LOGGER.warn("do not build sort config for groupId={}, as the
stream is empty", groupId);
```
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java:
##
@@ -96,4 +128,45 @@ public ListenerResult listen(WorkflowContext context)
throws WorkflowListenerExc
return ListenerResult.success("success");
}
+private void createQueueForStreams(InlongGroupInfo groupInfo,
List streamInfos, String operator) {
+String groupId = groupInfo.getInlongGroupId();
+log.info("success to start stream process for groupId={}", groupId);
+
+for (InlongStreamInfo stream : streamInfos) {
+StreamResourceProcessForm form =
StreamResourceProcessForm.getProcessForm(groupInfo, stream, INIT);
+String streamId = stream.getInlongStreamId();
+final String errMsg = "failed to start stream process for
groupId=" + groupId + " streamId=" + streamId;
+
+CompletableFuture future = CompletableFuture
+.supplyAsync(() ->
workflowService.start(CREATE_STREAM_RESOURCE, operator, form), EXECUTOR_SERVICE)
+.whenComplete((result, ex) -> {
+if (ex != null) {
+log.error(errMsg + ":
[GitHub] [inlong] leezng opened a new pull request, #6494: [INLONG-6493][Dashboard] After the Stream is configured successfully, some parameters cannot be modified
leezng opened a new pull request, #6494: URL: https://github.com/apache/inlong/pull/6494 - Fixes #6493 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #6046: [INLONG-6044][Manager] Distinguish between group and stream configuration processes
healchow commented on code in PR #6046:
URL: https://github.com/apache/inlong/pull/6046#discussion_r1018600385
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/sort/SortConfigListener.java:
##
@@ -81,6 +82,10 @@ public ListenerResult listen(WorkflowContext context) throws
WorkflowListenerExc
}
InlongGroupInfo groupInfo = form.getGroupInfo();
List streamInfos = form.getStreamInfos();
+if (CollectionUtils.isEmpty(streamInfos)) {
+LOGGER.warn("not build sort config for groupId={}, as the stream
is empty", groupId);
Review Comment:
Not necessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] fuweng11 commented on a diff in pull request #6046: [INLONG-6044][Manager] Distinguish between group and stream configuration processes
fuweng11 commented on code in PR #6046:
URL: https://github.com/apache/inlong/pull/6046#discussion_r1018602157
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/kafka/KafkaResourceOperators.java:
##
@@ -67,49 +62,12 @@ public boolean accept(String mqType) {
@Override
public void createQueueForGroup(@NotNull InlongGroupInfo groupInfo,
@NotBlank String operator) {
-String groupId = groupInfo.getInlongGroupId();
-log.info("begin to create kafka resource for groupId={}", groupId);
-
-InlongKafkaInfo inlongKafkaInfo = (InlongKafkaInfo) groupInfo;
-try {
-// 1. create kafka Topic - each Inlong Stream corresponds to a
Kafka Topic
-List streamInfoList =
streamService.getTopicList(groupId);
-if (streamInfoList == null || streamInfoList.isEmpty()) {
-log.warn("skip to create kafka topic and subscription as no
streams for groupId={}", groupId);
-return;
-}
-for (InlongStreamBriefInfo streamInfo : streamInfoList) {
-this.createKafkaTopic(inlongKafkaInfo,
streamInfo.getMqResource());
-}
-} catch (Exception e) {
-String msg = String.format("failed to create kafka resource for
groupId=%s", groupId);
-log.error(msg, e);
-throw new WorkflowListenerException(msg + ": " + e.getMessage());
-}
-log.info("success to create kafka resource for groupId={},
cluster={}", groupId, inlongKafkaInfo);
+log.info("skip to create kafka topic for groupId={}",
groupInfo.getInlongGroupId());
}
@Override
public void deleteQueueForGroup(InlongGroupInfo groupInfo, String
operator) {
-Preconditions.checkNotNull(groupInfo, "inlong group info cannot be
null");
-
-String groupId = groupInfo.getInlongGroupId();
-log.info("begin to delete kafka resource for groupId={}", groupId);
-ClusterInfo clusterInfo =
clusterService.getOne(groupInfo.getInlongClusterTag(), null, ClusterType.KAFKA);
-try {
-List streamInfoList =
streamService.getTopicList(groupId);
-if (streamInfoList == null || streamInfoList.isEmpty()) {
-log.warn("skip to create kafka topic and subscription as no
streams for groupId={}", groupId);
-return;
-}
-for (InlongStreamBriefInfo streamInfo : streamInfoList) {
-this.deleteKafkaTopic(groupInfo,
streamInfo.getInlongStreamId());
-}
-} catch (Exception e) {
-log.error("failed to delete kafka resource for groupId=" +
groupId, e);
-throw new WorkflowListenerException("failed to delete kafka
resource: " + e.getMessage());
-}
-log.info("success to delete kafka resource for groupId={},
cluster={}", groupId, clusterInfo);
+log.info("skip to delete kafka topic for groupId={}",
groupInfo.getInlongGroupId());
Review Comment:
I fixed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] fuweng11 commented on a diff in pull request #6046: [INLONG-6044][Manager] Distinguish between group and stream configuration processes
fuweng11 commented on code in PR #6046:
URL: https://github.com/apache/inlong/pull/6046#discussion_r1018604107
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java:
##
@@ -96,4 +128,45 @@ public ListenerResult listen(WorkflowContext context)
throws WorkflowListenerExc
return ListenerResult.success("success");
}
+private void createQueueForStreams(InlongGroupInfo groupInfo,
List streamInfos, String operator) {
+String groupId = groupInfo.getInlongGroupId();
+log.info("success to start stream process for groupId={}", groupId);
+
+for (InlongStreamInfo stream : streamInfos) {
+StreamResourceProcessForm form =
StreamResourceProcessForm.getProcessForm(groupInfo, stream, INIT);
+String streamId = stream.getInlongStreamId();
+final String errMsg = "failed to start stream process for
groupId=" + groupId + " streamId=" + streamId;
+
+CompletableFuture future = CompletableFuture
+.supplyAsync(() ->
workflowService.start(CREATE_STREAM_RESOURCE, operator, form), EXECUTOR_SERVICE)
+.whenComplete((result, ex) -> {
+if (ex != null) {
+log.error(errMsg + ": " + ex.getMessage());
+throw new WorkflowListenerException(errMsg, ex);
+} else {
+List tasks = result.getNewTasks();
+if (TaskStatus.FAILED == tasks.get(tasks.size() -
1).getStatus()) {
+log.error(errMsg);
+throw new WorkflowListenerException(errMsg);
+}
+}
+});
+try {
+future.get(180, TimeUnit.SECONDS);
+/*WorkflowResult result = future.get(180, TimeUnit.SECONDS);
+List tasks = result.getNewTasks();
+if (TaskStatus.FAILED == tasks.get(tasks.size() -
1).getStatus()) {
+log.error(errMsg);
+throw new WorkflowListenerException(errMsg);
+}*/
Review Comment:
I fixed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[inlong] branch master updated: [INLONG-6481][Dashboard] Supports management of SQLServer source (#6490)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new eed3c338e [INLONG-6481][Dashboard] Supports management of SQLServer
source (#6490)
eed3c338e is described below
commit eed3c338e5d1b01c498f1ce780114d96b24aad85
Author: Lizhen <88174078+bluew...@users.noreply.github.com>
AuthorDate: Thu Nov 10 10:55:18 2022 +0800
[INLONG-6481][Dashboard] Supports management of SQLServer source (#6490)
---
inlong-dashboard/src/locales/cn.json | 10 ++
inlong-dashboard/src/locales/en.json | 10 ++
.../src/metas/sources/defaults/SQLServer.ts| 151 +
.../src/metas/sources/defaults/index.ts| 5 +
4 files changed, 176 insertions(+)
diff --git a/inlong-dashboard/src/locales/cn.json
b/inlong-dashboard/src/locales/cn.json
index 44ad7e605..029da9daf 100644
--- a/inlong-dashboard/src/locales/cn.json
+++ b/inlong-dashboard/src/locales/cn.json
@@ -58,6 +58,16 @@
"meta.Sources.PostgreSQL.TableName": "表格名称",
"meta.Sources.PostgreSQL.decodingPluginName": "解码插件名称",
"meta.Sources.PostgreSQL.PrimaryKey": "主键",
+ "meta.Sources.SQLServer.Hostname": "服务器主机",
+ "meta.Sources.SQLServer.Port": "端口",
+ "meta.Sources.SQLServer.Username": "用户",
+ "meta.Sources.SQLServer.Password": "密码",
+ "meta.Sources.SQLServer.Database": "数据库名",
+ "meta.Sources.SQLServer.SchemaName": "架构名称",
+ "meta.Sources.SQLServer.TableName": "表格名称",
+ "meta.Sources.SQLServer.AllMigration": "是否整库迁移",
+ "meta.Sources.SQLServer.ServerTimezone": "服务器时区",
+ "meta.Sources.SQLServer.PrimaryKey": "主键",
"meta.Sinks.SinkName": "名称",
"meta.Sinks.SinkNameRule": "以英文字母开头，只能包含英文字母、数字、中划线、下划线",
"meta.Sinks.SinkType": "类型",
diff --git a/inlong-dashboard/src/locales/en.json
b/inlong-dashboard/src/locales/en.json
index ecdfcfa24..70e2e536e 100644
--- a/inlong-dashboard/src/locales/en.json
+++ b/inlong-dashboard/src/locales/en.json
@@ -58,6 +58,16 @@
"meta.Sources.PostgreSQL.TableName": "TableName",
"meta.Sources.PostgreSQL.decodingPluginName": "Decoding Plugin Name",
"meta.Sources.PostgreSQL.PrimaryKey": "PrimaryKey",
+ "meta.Sources.SQLServer.Hostname": "Hostname",
+ "meta.Sources.SQLServer.Port": "Port",
+ "meta.Sources.SQLServer.Username": "Username",
+ "meta.Sources.SQLServer.Password": "Password",
+ "meta.Sources.SQLServer.Database": "Database",
+ "meta.Sources.SQLServer.SchemaName": "SchemaName",
+ "meta.Sources.SQLServer.TableName": "TableName",
+ "meta.Sources.SQLServer.AllMigration": "AllMigration",
+ "meta.Sources.SQLServer.ServerTimezone": "Timezone",
+ "meta.Sources.SQLServer.PrimaryKey": "PrimaryKey",
"meta.Sinks.SinkName": "Name",
"meta.Sinks.SinkNameRule": "At the beginning of English letters, only
English letters, numbers, minus, and underscores",
"meta.Sinks.SinkType": "Type",
diff --git a/inlong-dashboard/src/metas/sources/defaults/SQLServer.ts
b/inlong-dashboard/src/metas/sources/defaults/SQLServer.ts
new file mode 100644
index 0..0bd6e40a8
--- /dev/null
+++ b/inlong-dashboard/src/metas/sources/defaults/SQLServer.ts
@@ -0,0 +1,151 @@
+/*
+ * 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.
+ */
+
+import { DataWithBackend } from '@/metas/DataWithBackend';
+import { RenderRow } from '@/metas/RenderRow';
+import { RenderList } from '@/metas/RenderList';
+import { SourceInfo } from '../common/SourceInfo';
+import i18n from '@/i18n';
+
+const { I18n } = DataWithBackend;
+const { FieldDecorator } = RenderRow;
+const { ColumnDecorator } = RenderList;
+
+export default class SQLServerSource
+ extends SourceInfo
+ implements DataWithBackend, RenderRow, RenderList
+{
+ @FieldDecorator({
+type: 'input',
+rules: [{ required: true }],
+props: values => ({
+ disabled: values?.status === 101,
+}),
+ })
+ @ColumnDecorator()
+ @I18n('meta.Sources.SQLServer.Hostname')
+ hostname: string;
+
+ @FieldDecorator({
+type: 'inputnumber',
+rules: [{ required: true }],
+initialValue: 1433,
+props: values => ({
+ min: 1,
+ max: 65535,
+ disabled: va
[GitHub] [inlong] dockerzhang merged pull request #6490: [INLONG-6481][Dashboard] Supports management of SQLServer source
dockerzhang merged PR #6490: URL: https://github.com/apache/inlong/pull/6490 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] dockerzhang merged pull request #6486: [INLONG-6484][Sort] Fix schema update circular dependency error in multiple sink iceberg scenes
dockerzhang merged PR #6486: URL: https://github.com/apache/inlong/pull/6486 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[inlong] branch master updated: [INLONG-6484][Sort] Bugfix: fix schema update Circular dependency error in multiple sink iceberg scenes (#6486)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 9aeb5f66c [INLONG-6484][Sort] Bugfix: fix schema update Circular
dependency error in multiple sink iceberg scenes (#6486)
9aeb5f66c is described below
commit 9aeb5f66c66b191bd31a5af376ddb9ba8dce2bc7
Author: thesumery <107393625+thesum...@users.noreply.github.com>
AuthorDate: Thu Nov 10 10:55:39 2022 +0800
[INLONG-6484][Sort] Bugfix: fix schema update Circular dependency error in
multiple sink iceberg scenes (#6486)
Co-authored-by: thesumery <158971...@qq.com>
---
.../inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java | 1 +
1 file changed, 1 insertion(+)
diff --git
a/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java
b/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java
index 378f040d9..ec65fbae7 100644
---
a/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java
+++
b/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java
@@ -186,6 +186,7 @@ public class DynamicSchemaHandleOperator extends
AbstractStreamOperator(recordWithSchema));
} else {
handldAlterSchemaEventFromOperator(tableId, latestSchema,
dataSchema);
+break;
}
}
}
[inlong] 02/03: [INLONG-6481][Dashboard] Supports management of SQLServer source (#6490)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.4
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 41d2854632f0c4b9fc236a41327bdaa363163e58
Author: Lizhen <88174078+bluew...@users.noreply.github.com>
AuthorDate: Thu Nov 10 10:55:18 2022 +0800
[INLONG-6481][Dashboard] Supports management of SQLServer source (#6490)
---
inlong-dashboard/src/locales/cn.json | 10 ++
inlong-dashboard/src/locales/en.json | 10 ++
.../src/metas/sources/defaults/SQLServer.ts| 151 +
.../src/metas/sources/defaults/index.ts| 5 +
4 files changed, 176 insertions(+)
diff --git a/inlong-dashboard/src/locales/cn.json
b/inlong-dashboard/src/locales/cn.json
index 44ad7e605..029da9daf 100644
--- a/inlong-dashboard/src/locales/cn.json
+++ b/inlong-dashboard/src/locales/cn.json
@@ -58,6 +58,16 @@
"meta.Sources.PostgreSQL.TableName": "表格名称",
"meta.Sources.PostgreSQL.decodingPluginName": "解码插件名称",
"meta.Sources.PostgreSQL.PrimaryKey": "主键",
+ "meta.Sources.SQLServer.Hostname": "服务器主机",
+ "meta.Sources.SQLServer.Port": "端口",
+ "meta.Sources.SQLServer.Username": "用户",
+ "meta.Sources.SQLServer.Password": "密码",
+ "meta.Sources.SQLServer.Database": "数据库名",
+ "meta.Sources.SQLServer.SchemaName": "架构名称",
+ "meta.Sources.SQLServer.TableName": "表格名称",
+ "meta.Sources.SQLServer.AllMigration": "是否整库迁移",
+ "meta.Sources.SQLServer.ServerTimezone": "服务器时区",
+ "meta.Sources.SQLServer.PrimaryKey": "主键",
"meta.Sinks.SinkName": "名称",
"meta.Sinks.SinkNameRule": "以英文字母开头，只能包含英文字母、数字、中划线、下划线",
"meta.Sinks.SinkType": "类型",
diff --git a/inlong-dashboard/src/locales/en.json
b/inlong-dashboard/src/locales/en.json
index ecdfcfa24..70e2e536e 100644
--- a/inlong-dashboard/src/locales/en.json
+++ b/inlong-dashboard/src/locales/en.json
@@ -58,6 +58,16 @@
"meta.Sources.PostgreSQL.TableName": "TableName",
"meta.Sources.PostgreSQL.decodingPluginName": "Decoding Plugin Name",
"meta.Sources.PostgreSQL.PrimaryKey": "PrimaryKey",
+ "meta.Sources.SQLServer.Hostname": "Hostname",
+ "meta.Sources.SQLServer.Port": "Port",
+ "meta.Sources.SQLServer.Username": "Username",
+ "meta.Sources.SQLServer.Password": "Password",
+ "meta.Sources.SQLServer.Database": "Database",
+ "meta.Sources.SQLServer.SchemaName": "SchemaName",
+ "meta.Sources.SQLServer.TableName": "TableName",
+ "meta.Sources.SQLServer.AllMigration": "AllMigration",
+ "meta.Sources.SQLServer.ServerTimezone": "Timezone",
+ "meta.Sources.SQLServer.PrimaryKey": "PrimaryKey",
"meta.Sinks.SinkName": "Name",
"meta.Sinks.SinkNameRule": "At the beginning of English letters, only
English letters, numbers, minus, and underscores",
"meta.Sinks.SinkType": "Type",
diff --git a/inlong-dashboard/src/metas/sources/defaults/SQLServer.ts
b/inlong-dashboard/src/metas/sources/defaults/SQLServer.ts
new file mode 100644
index 0..0bd6e40a8
--- /dev/null
+++ b/inlong-dashboard/src/metas/sources/defaults/SQLServer.ts
@@ -0,0 +1,151 @@
+/*
+ * 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.
+ */
+
+import { DataWithBackend } from '@/metas/DataWithBackend';
+import { RenderRow } from '@/metas/RenderRow';
+import { RenderList } from '@/metas/RenderList';
+import { SourceInfo } from '../common/SourceInfo';
+import i18n from '@/i18n';
+
+const { I18n } = DataWithBackend;
+const { FieldDecorator } = RenderRow;
+const { ColumnDecorator } = RenderList;
+
+export default class SQLServerSource
+ extends SourceInfo
+ implements DataWithBackend, RenderRow, RenderList
+{
+ @FieldDecorator({
+type: 'input',
+rules: [{ required: true }],
+props: values => ({
+ disabled: values?.status === 101,
+}),
+ })
+ @ColumnDecorator()
+ @I18n('meta.Sources.SQLServer.Hostname')
+ hostname: string;
+
+ @FieldDecorator({
+type: 'inputnumber',
+rules: [{ required: true }],
+initialValue: 1433,
+props: values => ({
+ min: 1,
+ max: 65535,
+ disabled: values?.status === 101,
+}),
+ })
+ @ColumnDecorator()
+ @I18n('meta.Sources.SQLServer.Port')
+ port: number;
+
+ @FieldDecorator({
+type: 'input',
+rules: [{ required: true }
[inlong] 03/03: [INLONG-6484][Sort] Bugfix: fix schema update Circular dependency error in multiple sink iceberg scenes (#6486)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.4
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 4634918b513ea84db6d06fa15d8b7200a521fd4b
Author: thesumery <107393625+thesum...@users.noreply.github.com>
AuthorDate: Thu Nov 10 10:55:39 2022 +0800
[INLONG-6484][Sort] Bugfix: fix schema update Circular dependency error in
multiple sink iceberg scenes (#6486)
Co-authored-by: thesumery <158971...@qq.com>
---
.../inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java | 1 +
1 file changed, 1 insertion(+)
diff --git
a/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java
b/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java
index 378f040d9..ec65fbae7 100644
---
a/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java
+++
b/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java
@@ -186,6 +186,7 @@ public class DynamicSchemaHandleOperator extends
AbstractStreamOperator(recordWithSchema));
} else {
handldAlterSchemaEventFromOperator(tableId, latestSchema,
dataSchema);
+break;
}
}
}
[inlong] 01/03: [INLONG-6491][Manager] Support getting backup info in getAllConfig (#6492)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch branch-1.4
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit f31a9d6da85f0251005115c610c698d35d93fd71
Author: 卢春亮 <946240...@qq.com>
AuthorDate: Thu Nov 10 10:00:09 2022 +0800
[INLONG-6491][Manager] Support getting backup info in getAllConfig (#6492)
---
.../repository/DataProxyConfigRepository.java | 340 +++--
1 file changed, 171 insertions(+), 169 deletions(-)
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/repository/DataProxyConfigRepository.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/repository/DataProxyConfigRepository.java
index d49eb9a34..55a5e7c85 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/repository/DataProxyConfigRepository.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/repository/DataProxyConfigRepository.java
@@ -19,10 +19,13 @@ package org.apache.inlong.manager.service.repository;
import com.google.common.base.Splitter;
import com.google.gson.Gson;
+import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
+
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.StringUtils;
+import org.apache.inlong.common.constant.ClusterSwitch;
import org.apache.inlong.common.pojo.dataproxy.CacheClusterObject;
import org.apache.inlong.common.pojo.dataproxy.CacheClusterSetObject;
import org.apache.inlong.common.pojo.dataproxy.DataProxyCluster;
@@ -34,20 +37,23 @@ import
org.apache.inlong.common.pojo.dataproxy.RepositoryTimerTask;
import org.apache.inlong.manager.common.consts.InlongConstants;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.exceptions.BusinessException;
-import org.apache.inlong.manager.pojo.cluster.ClusterPageRequest;
-import org.apache.inlong.manager.pojo.dataproxy.CacheCluster;
-import org.apache.inlong.manager.pojo.dataproxy.InlongGroupId;
-import org.apache.inlong.manager.pojo.dataproxy.InlongStreamId;
-import org.apache.inlong.manager.pojo.dataproxy.ProxyCluster;
-import org.apache.inlong.manager.pojo.sink.SinkPageRequest;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.dao.entity.InlongClusterEntity;
import org.apache.inlong.manager.dao.entity.InlongGroupEntity;
+import org.apache.inlong.manager.dao.entity.InlongGroupExtEntity;
+import org.apache.inlong.manager.dao.entity.InlongStreamExtEntity;
import org.apache.inlong.manager.dao.entity.StreamSinkEntity;
import org.apache.inlong.manager.dao.mapper.ClusterSetMapper;
import org.apache.inlong.manager.dao.mapper.InlongClusterEntityMapper;
import org.apache.inlong.manager.dao.mapper.InlongGroupEntityMapper;
import org.apache.inlong.manager.dao.mapper.StreamSinkEntityMapper;
+import org.apache.inlong.manager.pojo.cluster.ClusterPageRequest;
+import org.apache.inlong.manager.pojo.dataproxy.CacheCluster;
+import org.apache.inlong.manager.pojo.dataproxy.InlongGroupId;
+import org.apache.inlong.manager.pojo.dataproxy.InlongStreamId;
+import org.apache.inlong.manager.pojo.dataproxy.ProxyCluster;
+import org.apache.inlong.manager.pojo.sink.SinkPageRequest;
+import org.apache.inlong.manager.service.core.SortConfigLoader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -55,7 +61,6 @@ import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Transactional;
-import javax.annotation.PostConstruct;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Date;
@@ -63,10 +68,13 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Optional;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.PostConstruct;
+
/**
* DataProxyConfigRepository
*/
@@ -76,6 +84,7 @@ public class DataProxyConfigRepository implements IRepository
{
public static final Logger LOGGER =
LoggerFactory.getLogger(DataProxyConfigRepository.class);
+public static final String KEY_NAMESPACE = "namespace";
public static final String KEY_BACKUP_CLUSTER_TAG = "backup_cluster_tag";
public static final String KEY_BACKUP_TOPIC = "backup_topic";
public static final String KEY_SORT_TASK_NAME = "defaultSortTaskName";
@@ -104,6 +113,8 @@ public class DataProxyConfigRepository implements
IRepository {
private InlongGroupEntityMapper inlongGroupMapper;
@Autowired
private StreamSinkEntityMapper streamSinkMapper;
+@Autowired
+private SortConfigL
[inlong] branch branch-1.4 updated (287644e7c -> 4634918b5)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a change to branch branch-1.4
in repository https://gitbox.apache.org/repos/asf/inlong.git
from 287644e7c [INLONG-6406][DataProxy] Should support creating sink
dynamically after started (addendum) (#6488)
new f31a9d6da [INLONG-6491][Manager] Support getting backup info in
getAllConfig (#6492)
new 41d285463 [INLONG-6481][Dashboard] Supports management of SQLServer
source (#6490)
new 4634918b5 [INLONG-6484][Sort] Bugfix: fix schema update Circular
dependency error in multiple sink iceberg scenes (#6486)
The 3 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 | 10 +
inlong-dashboard/src/locales/en.json | 10 +
.../sources/defaults/{Oracle.ts => SQLServer.ts} | 72 ++---
.../src/metas/sources/defaults/index.ts| 5 +
.../repository/DataProxyConfigRepository.java | 340 +++--
.../sink/multiple/DynamicSchemaHandleOperator.java | 1 +
6 files changed, 228 insertions(+), 210 deletions(-)
copy inlong-dashboard/src/metas/sources/defaults/{Oracle.ts => SQLServer.ts}
(81%)
[GitHub] [inlong] chestnut-c commented on a diff in pull request #6046: [INLONG-6044][Manager] Distinguish between group and stream configuration processes
chestnut-c commented on code in PR #6046:
URL: https://github.com/apache/inlong/pull/6046#discussion_r1018684017
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java:
##
@@ -96,4 +128,39 @@ public ListenerResult listen(WorkflowContext context)
throws WorkflowListenerExc
return ListenerResult.success("success");
}
+private void createQueueForStreams(InlongGroupInfo groupInfo,
List streamInfos, String operator) {
+String groupId = groupInfo.getInlongGroupId();
+log.info("success to start stream process for groupId={}", groupId);
+
+for (InlongStreamInfo stream : streamInfos) {
+StreamResourceProcessForm form =
StreamResourceProcessForm.getProcessForm(groupInfo, stream, INIT);
+String streamId = stream.getInlongStreamId();
+final String errMsg = "failed to start stream process for
groupId=" + groupId + " streamId=" + streamId;
+
+CompletableFuture future = CompletableFuture
+.supplyAsync(() ->
workflowService.start(CREATE_STREAM_RESOURCE, operator, form), EXECUTOR_SERVICE)
+.whenComplete((result, ex) -> {
+if (ex != null) {
+log.error(errMsg + ": " + ex.getMessage());
+throw new WorkflowListenerException(errMsg, ex);
+} else {
+List tasks = result.getNewTasks();
+if (TaskStatus.FAILED == tasks.get(tasks.size() -
1).getStatus()) {
+log.error(errMsg);
+throw new WorkflowListenerException(errMsg);
+}
+}
+});
+try {
+future.get(180, TimeUnit.SECONDS);
Review Comment:
Waiting for the result in the for loop is equivalent to synchronization. Why
use the asynchronous method?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] fuweng11 commented on a diff in pull request #6046: [INLONG-6044][Manager] Distinguish between group and stream configuration processes
fuweng11 commented on code in PR #6046:
URL: https://github.com/apache/inlong/pull/6046#discussion_r1018691530
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java:
##
@@ -96,4 +128,39 @@ public ListenerResult listen(WorkflowContext context)
throws WorkflowListenerExc
return ListenerResult.success("success");
}
+private void createQueueForStreams(InlongGroupInfo groupInfo,
List streamInfos, String operator) {
+String groupId = groupInfo.getInlongGroupId();
+log.info("success to start stream process for groupId={}", groupId);
+
+for (InlongStreamInfo stream : streamInfos) {
+StreamResourceProcessForm form =
StreamResourceProcessForm.getProcessForm(groupInfo, stream, INIT);
+String streamId = stream.getInlongStreamId();
+final String errMsg = "failed to start stream process for
groupId=" + groupId + " streamId=" + streamId;
+
+CompletableFuture future = CompletableFuture
+.supplyAsync(() ->
workflowService.start(CREATE_STREAM_RESOURCE, operator, form), EXECUTOR_SERVICE)
+.whenComplete((result, ex) -> {
+if (ex != null) {
+log.error(errMsg + ": " + ex.getMessage());
+throw new WorkflowListenerException(errMsg, ex);
+} else {
+List tasks = result.getNewTasks();
+if (TaskStatus.FAILED == tasks.get(tasks.size() -
1).getStatus()) {
+log.error(errMsg);
+throw new WorkflowListenerException(errMsg);
+}
+}
+});
+try {
+future.get(180, TimeUnit.SECONDS);
Review Comment:
Here we need to wait for the stream configuration to complete before
configuring the next stream. Otherwise, an exception is thrown and the next
stream configuration is not performed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] chestnut-c commented on a diff in pull request #6046: [INLONG-6044][Manager] Distinguish between group and stream configuration processes
chestnut-c commented on code in PR #6046:
URL: https://github.com/apache/inlong/pull/6046#discussion_r1018693753
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java:
##
@@ -96,4 +128,39 @@ public ListenerResult listen(WorkflowContext context)
throws WorkflowListenerExc
return ListenerResult.success("success");
}
+private void createQueueForStreams(InlongGroupInfo groupInfo,
List streamInfos, String operator) {
+String groupId = groupInfo.getInlongGroupId();
+log.info("success to start stream process for groupId={}", groupId);
+
+for (InlongStreamInfo stream : streamInfos) {
+StreamResourceProcessForm form =
StreamResourceProcessForm.getProcessForm(groupInfo, stream, INIT);
+String streamId = stream.getInlongStreamId();
+final String errMsg = "failed to start stream process for
groupId=" + groupId + " streamId=" + streamId;
+
+CompletableFuture future = CompletableFuture
+.supplyAsync(() ->
workflowService.start(CREATE_STREAM_RESOURCE, operator, form), EXECUTOR_SERVICE)
+.whenComplete((result, ex) -> {
+if (ex != null) {
+log.error(errMsg + ": " + ex.getMessage());
+throw new WorkflowListenerException(errMsg, ex);
+} else {
+List tasks = result.getNewTasks();
+if (TaskStatus.FAILED == tasks.get(tasks.size() -
1).getStatus()) {
+log.error(errMsg);
+throw new WorkflowListenerException(errMsg);
+}
+}
+});
+try {
+future.get(180, TimeUnit.SECONDS);
Review Comment:
Why not just use the synchronous method?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] haifxu commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
haifxu commented on code in PR #6489:
URL: https://github.com/apache/inlong/pull/6489#discussion_r1018696821
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java:
##
@@ -267,6 +267,21 @@ public Boolean delete(Integer id, String operator) {
return true;
}
+@Override
+public Boolean forceDelete(String groupId, String streamId, String
operator) {
+LOGGER.info("begin to force delete source for groupId={} and
streamId={} by user={}",
+groupId, streamId, operator);
+Preconditions.checkNotNull(groupId,
ErrorCodeEnum.GROUP_ID_IS_EMPTY.getMessage());
+Preconditions.checkNotNull(streamId,
ErrorCodeEnum.STREAM_ID_IS_EMPTY.getMessage());
+
+sourceMapper.updateByRelatedId(groupId, streamId,
SourceStatus.SOURCE_DISABLE.getCode());
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] fuweng11 commented on a diff in pull request #6046: [INLONG-6044][Manager] Distinguish between group and stream configuration processes
fuweng11 commented on code in PR #6046:
URL: https://github.com/apache/inlong/pull/6046#discussion_r1018697052
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java:
##
@@ -96,4 +128,39 @@ public ListenerResult listen(WorkflowContext context)
throws WorkflowListenerExc
return ListenerResult.success("success");
}
+private void createQueueForStreams(InlongGroupInfo groupInfo,
List streamInfos, String operator) {
+String groupId = groupInfo.getInlongGroupId();
+log.info("success to start stream process for groupId={}", groupId);
+
+for (InlongStreamInfo stream : streamInfos) {
+StreamResourceProcessForm form =
StreamResourceProcessForm.getProcessForm(groupInfo, stream, INIT);
+String streamId = stream.getInlongStreamId();
+final String errMsg = "failed to start stream process for
groupId=" + groupId + " streamId=" + streamId;
+
+CompletableFuture future = CompletableFuture
+.supplyAsync(() ->
workflowService.start(CREATE_STREAM_RESOURCE, operator, form), EXECUTOR_SERVICE)
+.whenComplete((result, ex) -> {
+if (ex != null) {
+log.error(errMsg + ": " + ex.getMessage());
+throw new WorkflowListenerException(errMsg, ex);
+} else {
+List tasks = result.getNewTasks();
+if (TaskStatus.FAILED == tasks.get(tasks.size() -
1).getStatus()) {
+log.error(errMsg);
+throw new WorkflowListenerException(errMsg);
+}
+}
+});
+try {
+future.get(180, TimeUnit.SECONDS);
Review Comment:
This is because you need to call the stream workflow to configure. If you
add a stream later, you do not need to configure the group workflow, but
directly execute the stream workflow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] fuweng11 commented on a diff in pull request #6046: [INLONG-6044][Manager] Distinguish between group and stream configuration processes
fuweng11 commented on code in PR #6046:
URL: https://github.com/apache/inlong/pull/6046#discussion_r1018697052
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java:
##
@@ -96,4 +128,39 @@ public ListenerResult listen(WorkflowContext context)
throws WorkflowListenerExc
return ListenerResult.success("success");
}
+private void createQueueForStreams(InlongGroupInfo groupInfo,
List streamInfos, String operator) {
+String groupId = groupInfo.getInlongGroupId();
+log.info("success to start stream process for groupId={}", groupId);
+
+for (InlongStreamInfo stream : streamInfos) {
+StreamResourceProcessForm form =
StreamResourceProcessForm.getProcessForm(groupInfo, stream, INIT);
+String streamId = stream.getInlongStreamId();
+final String errMsg = "failed to start stream process for
groupId=" + groupId + " streamId=" + streamId;
+
+CompletableFuture future = CompletableFuture
+.supplyAsync(() ->
workflowService.start(CREATE_STREAM_RESOURCE, operator, form), EXECUTOR_SERVICE)
+.whenComplete((result, ex) -> {
+if (ex != null) {
+log.error(errMsg + ": " + ex.getMessage());
+throw new WorkflowListenerException(errMsg, ex);
+} else {
+List tasks = result.getNewTasks();
+if (TaskStatus.FAILED == tasks.get(tasks.size() -
1).getStatus()) {
+log.error(errMsg);
+throw new WorkflowListenerException(errMsg);
+}
+}
+});
+try {
+future.get(180, TimeUnit.SECONDS);
Review Comment:
This is because we need to call the stream workflow to configure. If we add
a stream later, we do not need to configure the group workflow, but directly
execute the stream workflow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] fuweng11 commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
fuweng11 commented on code in PR #6489:
URL: https://github.com/apache/inlong/pull/6489#discussion_r1018700609
##
inlong-manager/manager-dao/src/main/resources/mappers/StreamSourceFieldEntityMapper.xml:
##
@@ -258,6 +258,15 @@
from stream_source_field
where source_id = #{sourceId,jdbcType=INTEGER}
+
Review Comment:
Suggest `deleteByRelatedId` -> `updateByRelatedId`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
healchow commented on code in PR #6489:
URL: https://github.com/apache/inlong/pull/6489#discussion_r1018700988
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java:
##
@@ -267,6 +267,20 @@ public Boolean delete(Integer id, String operator) {
return true;
}
+@Override
+public Boolean forceDelete(String groupId, String streamId, String
operator) {
+LOGGER.info("begin to force delete source for groupId={} and
streamId={} by user={}",
+groupId, streamId, operator);
+Preconditions.checkNotNull(groupId,
ErrorCodeEnum.GROUP_ID_IS_EMPTY.getMessage());
+Preconditions.checkNotNull(streamId,
ErrorCodeEnum.STREAM_ID_IS_EMPTY.getMessage());
+
+int sourceCount = sourceMapper.updateByRelatedId(groupId, streamId,
SourceStatus.SOURCE_DISABLE.getCode());
+int fieldCount = sourceFieldMapper.deleteByRelatedId(groupId,
streamId);
+LOGGER.info("success to force delete source for groupId={} and
streamId={} by user={}, update {} sources and delete {} fields",
+groupId, streamId, operator, sourceCount, fieldCount);
Review Comment:
```suggestion
LOGGER.info("success to force delete source for groupId={} and
streamId={} by user={}, update {} sources and {} fields",
groupId, streamId, operator, sourceCount, fieldCount);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] haifxu commented on a diff in pull request #6489: [INLONG-6487][Manager] Add API to force delete the stream source
haifxu commented on code in PR #6489:
URL: https://github.com/apache/inlong/pull/6489#discussion_r1018705825
##
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java:
##
@@ -267,6 +267,20 @@ public Boolean delete(Integer id, String operator) {
return true;
}
+@Override
+public Boolean forceDelete(String groupId, String streamId, String
operator) {
+LOGGER.info("begin to force delete source for groupId={} and
streamId={} by user={}",
+groupId, streamId, operator);
+Preconditions.checkNotNull(groupId,
ErrorCodeEnum.GROUP_ID_IS_EMPTY.getMessage());
+Preconditions.checkNotNull(streamId,
ErrorCodeEnum.STREAM_ID_IS_EMPTY.getMessage());
+
+int sourceCount = sourceMapper.updateByRelatedId(groupId, streamId,
SourceStatus.SOURCE_DISABLE.getCode());
+int fieldCount = sourceFieldMapper.deleteByRelatedId(groupId,
streamId);
+LOGGER.info("success to force delete source for groupId={} and
streamId={} by user={}, update {} sources and delete {} fields",
+groupId, streamId, operator, sourceCount, fieldCount);
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong] healchow merged pull request #6046: [INLONG-6044][Manager] Distinguish between group and stream configuration processes
healchow merged PR #6046: URL: https://github.com/apache/inlong/pull/6046 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[inlong] branch master updated: [INLONG-6044][Manager] Distinguish config processes between the InlongGroup and InlongStream (#6046)
This is an automated email from the ASF dual-hosted git repository.
healchow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new c21437090 [INLONG-6044][Manager] Distinguish config processes between
the InlongGroup and InlongStream (#6046)
c21437090 is described below
commit c2143709048cc2a7a6f5dc50ea92575042654182
Author: fuweng11 <76141879+fuwen...@users.noreply.github.com>
AuthorDate: Thu Nov 10 14:46:44 2022 +0800
[INLONG-6044][Manager] Distinguish config processes between the InlongGroup
and InlongStream (#6046)
* Distinguish config processes between the InlongGroup and InlongStream
* Use common thread pool for UpdateGroupCompleteListener
* Modify the source state according to lightweight
* Modify the asynchronous process logic of inlong stream
* Allow configuration without inlong stream
* Determine whether to configure Sort according to isStream param
Co-authored-by: healchow
---
.../manager/dao/entity/WorkflowProcessEntity.java | 1 +
.../mappers/WorkflowProcessEntityMapper.xml| 27 +
.../manager/pojo/workflow/ProcessRequest.java | 3 +
.../manager/pojo/workflow/TaskLogRequest.java | 3 +
.../form/process/StreamResourceProcessForm.java| 13 +
.../listener/group/InitGroupCompleteListener.java | 24 +++-
.../service/listener/group/InitGroupListener.java | 4 --
.../listener/queue/QueueResourceListener.java | 67 ++
.../queue/StreamQueueResourceListener.java | 3 -
.../service/listener/sort/SortConfigListener.java | 5 ++
.../stream/InitStreamCompleteListener.java | 7 ++-
.../queue/kafka/KafkaResourceOperators.java| 25 +---
.../queue/pulsar/PulsarResourceOperator.java | 16 +-
.../resource/sort/DefaultSortConfigOperator.java | 10 ++--
.../service/stream/InlongStreamProcessService.java | 20 +++
.../service/workflow/WorkflowServiceImpl.java | 1 +
.../group/CreateGroupWorkflowDefinition.java | 24 +---
.../group/DeleteGroupWorkflowDefinition.java | 8 ++-
.../stream/CreateStreamWorkflowDefinition.java | 4 +-
.../stream/DeleteStreamWorkflowDefinition.java | 4 +-
.../group/CreateGroupWorkflowDefinitionTest.java | 4 +-
.../main/resources/h2/apache_inlong_manager.sql| 3 +-
.../manager-web/sql/apache_inlong_manager.sql | 3 +-
.../workflow/processor/StartEventProcessor.java| 5 ++
24 files changed, 162 insertions(+), 122 deletions(-)
diff --git
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/WorkflowProcessEntity.java
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/WorkflowProcessEntity.java
index 87e92bf4a..b295e1e40 100644
---
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/WorkflowProcessEntity.java
+++
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/WorkflowProcessEntity.java
@@ -34,6 +34,7 @@ public class WorkflowProcessEntity {
private String title;
private String inlongGroupId;
+private String inlongStreamId;
private String applicant;
private String status;
private String formData;
diff --git
a/inlong-manager/manager-dao/src/main/resources/mappers/WorkflowProcessEntityMapper.xml
b/inlong-manager/manager-dao/src/main/resources/mappers/WorkflowProcessEntityMapper.xml
index 1969918b4..8b122ab12 100644
---
a/inlong-manager/manager-dao/src/main/resources/mappers/WorkflowProcessEntityMapper.xml
+++
b/inlong-manager/manager-dao/src/main/resources/mappers/WorkflowProcessEntityMapper.xml
@@ -27,6 +27,7 @@
+
@@ -36,22 +37,21 @@
-id, name, display_name, type, title, inlong_group_id, applicant,
-status, start_time, end_time, hidden, form_data, ext_params
+id, name, display_name, type, title, inlong_group_id, inlong_stream_id,
+applicant, status, start_time, end_time, hidden, form_data, ext_params
-insert into workflow_process (name, display_name,
- type, title, inlong_group_id,
- applicant, status,
- start_time, end_time,
- form_data, ext_params, hidden)
-values (#{name,jdbcType=VARCHAR}, #{displayName,jdbcType=VARCHAR},
-#{type,jdbcType=VARCHAR}, #{title,jdbcType=VARCHAR},
#{inlongGroupId,jdbcType=VARCHAR},
-#{applicant,jdbcType=VARCHAR}, #{status,jdbcType=VARCHAR},
-#{startTime,jdbcType=TIMESTAMP}, #{endTime,jdbcType=TIMESTAMP},
-#{formData,jdbcType=LONGVARCHAR},
#{extParams,jdbcType=LONGVARCHAR}, #{hidden,jdbcType=TIN
[GitHub] [inlong] leezng merged pull request #6494: [INLONG-6493][Dashboard] After the Stream is configured successfully, some parameters cannot be modified
leezng merged PR #6494: URL: https://github.com/apache/inlong/pull/6494 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[inlong] branch master updated: [INLONG-6493][Dashboard] After the Stream is configured successfully, some parameters cannot be modified (#6494)
This is an automated email from the ASF dual-hosted git repository.
leezng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new dbddbbba0 [INLONG-6493][Dashboard] After the Stream is configured
successfully, some parameters cannot be modified (#6494)
dbddbbba0 is described below
commit dbddbbba08d420ecaaa3bb4a7a9a8c66d42b28f2
Author: Daniel
AuthorDate: Thu Nov 10 15:10:01 2022 +0800
[INLONG-6493][Dashboard] After the Stream is configured successfully, some
parameters cannot be modified (#6494)
---
.../src/metas/streams/common/StreamDefaultInfo.ts | 35 ++
.../GroupDetail/DataStream/StreamItemModal.tsx | 14 ++---
2 files changed, 25 insertions(+), 24 deletions(-)
diff --git a/inlong-dashboard/src/metas/streams/common/StreamDefaultInfo.ts
b/inlong-dashboard/src/metas/streams/common/StreamDefaultInfo.ts
index 773a51b19..fd39a8cc9 100644
--- a/inlong-dashboard/src/metas/streams/common/StreamDefaultInfo.ts
+++ b/inlong-dashboard/src/metas/streams/common/StreamDefaultInfo.ts
@@ -38,9 +38,10 @@ export class StreamDefaultInfo implements DataWithBackend,
RenderRow, RenderList
@FieldDecorator({
type: 'input',
-props: {
- maxLength: 32,
-},
+props: values => ({
+ disabled: Boolean(values?.status),
+ maxLength: 64,
+}),
rules: [
{ required: true },
{
@@ -97,7 +98,8 @@ export class StreamDefaultInfo implements DataWithBackend,
RenderRow, RenderList
type: 'radio',
initialValue: 'CSV',
tooltip: i18n.t('meta.Stream.DataTypeHelp'),
-props: {
+props: values => ({
+ disabled: [110, 130].includes(values?.status),
options: [
{
label: 'CSV',
@@ -108,7 +110,7 @@ export class StreamDefaultInfo implements DataWithBackend,
RenderRow, RenderList
value: 'RAW_CSV',
},
],
-},
+}),
rules: [{ required: true }],
})
@I18n('meta.Stream.DataType')
@@ -117,7 +119,8 @@ export class StreamDefaultInfo implements DataWithBackend,
RenderRow, RenderList
@FieldDecorator({
type: 'radio',
initialValue: 'UTF-8',
-props: {
+props: values => ({
+ disabled: [110, 130].includes(values?.status),
options: [
{
label: 'UTF-8',
@@ -128,7 +131,7 @@ export class StreamDefaultInfo implements DataWithBackend,
RenderRow, RenderList
value: 'GBK',
},
],
-},
+}),
rules: [{ required: true }],
})
@I18n('meta.Stream.DataEncoding')
@@ -137,7 +140,8 @@ export class StreamDefaultInfo implements DataWithBackend,
RenderRow, RenderList
@FieldDecorator({
type: 'select',
initialValue: '124',
-props: {
+props: values => ({
+ disabled: [110, 130].includes(values?.status),
dropdownMatchSelectWidth: false,
options: [
{
@@ -170,7 +174,7 @@ export class StreamDefaultInfo implements DataWithBackend,
RenderRow, RenderList
placeholder: 'ASCII',
},
style: { width: 100 },
-},
+}),
rules: [
{
required: true,
@@ -186,8 +190,9 @@ export class StreamDefaultInfo implements DataWithBackend,
RenderRow, RenderList
@FieldDecorator({
type: EditableTable,
-props: {
+props: values => ({
size: 'small',
+ canDelete: record => !(record.id && [110, 130].includes(values?.status)),
columns: [
{
title: i18n.t('meta.Stream.FieldName'),
@@ -199,15 +204,19 @@ export class StreamDefaultInfo implements
DataWithBackend, RenderRow, RenderList
message: i18n.t('meta.Stream.FieldNameRule'),
},
],
+ props: (text, record) => ({
+disabled: record.id && [110, 130].includes(values?.status),
+ }),
},
{
title: i18n.t('meta.Stream.FieldType'),
dataIndex: 'fieldType',
type: 'select',
initialValue: sourceFieldsTypes[0].value,
- props: {
+ props: (text, record) => ({
+disabled: record.id && [110, 130].includes(values?.status),
options: sourceFieldsTypes,
- },
+ }),
rules: [{ required: true }],
},
{
@@ -215,7 +224,7 @@ export class StreamDefaultInfo implements DataWithBackend,
RenderRow, RenderList
dataIndex: 'fieldComment',
},
],
-},
+}),
})
@I18n('meta.Stream.SourceDataField')
rowTypeFields: Record[];
diff --git
a/inlong-dashboard/src/pages/GroupDetail/DataStream/StreamItemModal.tsx
b/inlong-dashboard/src/pages/GroupDetail/DataStream/StreamItemModal.tsx
index 08c4af720..2c78e0e31 100644
--- a/inlong-dashboard/src/pages/GroupDetail/DataStream/StreamItemModal.tsx
+++ b/inlong-dashboard/src/pages/GroupDetail/DataStream/StreamItemModal.tsx
@@ -96,17 +96,8 @@ const Comp: React.FC = ({ inlon
[GitHub] [inlong] leezng opened a new pull request, #6499: [INLONG-6498][Dashboard] Field Resolution for Unified Group Approval
leezng opened a new pull request, #6499: URL: https://github.com/apache/inlong/pull/6499 - Fixes #6498 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong-website] thesumery opened a new issue, #591: [Feature][Doc] Add document for multiple sink of iceberg
thesumery opened a new issue, #591: URL: https://github.com/apache/inlong-website/issues/591 ### Description _No response_ ### Use case _No response_ ### Are you willing to submit PR? - [X] Yes, I am willing to submit a PR! ### Code of Conduct - [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong] vernedeng opened a new pull request, #6501: [INLONG-6497][Manager] Support Elasticsearch cluster
vernedeng opened a new pull request, #6501: URL: https://github.com/apache/inlong/pull/6501 - Fixes #6497 ### Motivation Support Elasticsearch cluster ### Modifications *Describe the modifications you've done.* ### Verifying this change *(Please pick either of the following options)* - [ ] This change is a trivial rework/code cleanup without any test coverage. ### Documentation - Does this pull request introduce a new feature? yes - If yes, how is the feature documented? JavaDocs -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong-website] thesumery opened a new pull request, #592: [INLONG-591][Doc] Add document for multiple sink of iceberg
thesumery opened a new pull request, #592: URL: https://github.com/apache/inlong-website/pull/592 [INLONG-591][Doc] Add document for multiple sink of iceberg ### Prepare a Pull Request - Add document for multiple sink of iceberg - Fixes #591 ### Motivation *Add document for multiple sink of iceberg.* ### Modifications *Add document for multiple sink of iceberg.* -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong-website] EMsnap commented on a diff in pull request #592: [INLONG-591][Doc] Add document for multiple sink of iceberg
EMsnap commented on code in PR #592:
URL: https://github.com/apache/inlong-website/pull/592#discussion_r1018738270
##
i18n/zh-CN/docusaurus-plugin-content-docs/current/data_node/load_node/iceberg.md:
##
@@ -148,6 +148,98 @@ TODO
### InLong Manager Client 用法
TODO
+## 特征
+### 多表写入
+目前 Iceberg 支持多表同时写入，需要在 FLINK SQL 的建表参数上添加 `'sink.multiple.enable' = 'true'`
并且目标表的schema
+只能定义成 `BINARY` 或者 `STRING` ，以下是一个建表语句举例：
+```
+CREATE TABLE `table_2`(
+`data` STRING)
+WITH (
+'connector'='iceberg-inlong',
+'catalog-name'='hive_prod',
+'uri'='thrift://localhost:9083',
+'warehouse'='hdfs://localhost:8020/hive/warehouse',
+'sink.multiple.enable' = 'true',
+'sink.multiple.format' = 'canal-json',
+'sink.multiple.add-column.policy' = 'TRY_IT_BEST',
+'sink.multiple.database-pattern' = '${database}',
+'sink.multiple.table-pattern' = 'test_${table}'
+);
+```
+要支持多表写入同时需要设置上游数据的序列化格式(通过选项 'sink.multiple.format'
+来设置, 目前仅支持 [canal-json|debezium-json])。
+
+### 动态表名映射
+Iceberg 在多表写入的时可以自定义映射的数据库名和表名的规则，可以填充占位符然后添加前后缀来修改映射的目标表名称。
+Iceberg Load Node 会解析 `'sink.multiple.database-pattern'` 作为目的端的 数据库名, 解析
`'sink.multiple.table-pattern'`
+作为目的端的表名，占位符是从数据中解析出来的，变量是严格通过 '${VARIABLE_NAME}' 来表示, 变量的取值来自于数据本身,
+即可以是通过 `'sink.multiple.format'` 指定的某种 Format 的元数据字段, 也可以是数据中的物理字段。
+关于 'topic-parttern' 的例子如下:
+- 'sink.multiple.format' 为 'canal-json':
+
+上游数据为:
+```
+{
+ "data": [
+{
+ "id": "111",
+ "name": "scooter",
+ "description": "Big 2-wheel scooter",
+ "weight": "5.18"
+}
+ ],
+ "database": "inventory",
+ "es": 158937356,
+ "id": 9,
+ "isDdl": false,
+ "mysqlType": {
+"id": "INTEGER",
+"name": "VARCHAR(255)",
+"description": "VARCHAR(512)",
+"weight": "FLOAT"
+ },
+ "old": [
+{
+ "weight": "5.15"
+}
+ ],
+ "pkNames": [
+"id"
+ ],
+ "sql": "",
+ "sqlType": {
+"id": 4,
+"name": 12,
+"description": 12,
+"weight": 7
+ },
+ "table": "products",
+ "ts": 1589373560798,
+ "type": "UPDATE"
+}
+```
+'topic-pattern' 为 '{database}_${table}', 提取后的 Topic 为 'inventory_products'
('database', 'table' 为元数据字段,
+'id' 为物理字段)
+
+'topic-pattern' 为 '{database}_${table}_${id}', 提取后的 Topic 为
'inventory_products_4' ('database', 'table'
Review Comment:
111 ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [inlong-website] gong commented on a diff in pull request #592: [INLONG-591][Doc] Add document for multiple sink of iceberg
gong commented on code in PR #592: URL: https://github.com/apache/inlong-website/pull/592#discussion_r1018748669 ## docs/data_node/load_node/iceberg.md: ## @@ -147,6 +147,103 @@ TODO ### Usage for InLong Manager Client TODO +## Feature +### Multiple table sink +Currently Iceberg support multiple table sinking, it require FLINK SQL create table parameters add +`'sink.multiple.enable' = 'true'` and target table schema can only be defined as `BINARY` or `STRING` Review Comment: `BINARY` should change to `BYTES` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong-website] gong commented on a diff in pull request #592: [INLONG-591][Doc] Add document for multiple sink of iceberg
gong commented on code in PR #592: URL: https://github.com/apache/inlong-website/pull/592#discussion_r1018751143 ## i18n/zh-CN/docusaurus-plugin-content-docs/current/data_node/load_node/iceberg.md: ## @@ -148,6 +148,98 @@ TODO ### InLong Manager Client 用法 TODO +## 特征 +### 多表写入 +目前 Iceberg 支持多表同时写入，需要在 FLINK SQL 的建表参数上添加 `'sink.multiple.enable' = 'true'` 并且目标表的schema +只能定义成 `BINARY` 或者 `STRING` ，以下是一个建表语句举例： +``` Review Comment: `BINARY` should be `BYTES` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [inlong-website] leezng closed issue #576: [Improve] Optimize document style layout
leezng closed issue #576: [Improve] Optimize document style layout URL: https://github.com/apache/inlong-website/issues/576 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
