Skip to content

Commit f42939f

Browse files
baisui1981baisui1981
committed
add kingbase KingbaseTableSinkFactory for SQL stype script supporting ,datavane/tis#465 
1 parent 84a5ab0 commit f42939f

16 files changed

Lines changed: 134 additions & 57 deletions

File tree

‎tis-datax/executor/dolphinscheduler-task-tis-datasync/dependency-reduced-pom.xml‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3,13 +3,13 @@
33
<parent>
44
<artifactId>tis-datax</artifactId>
55
<groupId>com.qlangtech.tis.plugins</groupId>
6-
<version>4.3.0-SNAPSHOT</version>
6+
<version>4.3.0</version>
77
<relativePath>../../pom.xml</relativePath>
88
</parent>
99
<modelVersion>4.0.0</modelVersion>
1010
<groupId>com.qlangtech.tis.plugins</groupId>
1111
<artifactId>dolphinscheduler-task-tis-datasync</artifactId>
12-
<version>4.3.0-SNAPSHOT</version>
12+
<version>4.3.0</version>
1313
<licenses>
1414
<license>
1515
<name>GNU Affero General Public License</name>
@@ -68,7 +68,7 @@
6868
<dependency>
6969
<groupId>com.qlangtech.tis</groupId>
7070
<artifactId>tis-base-test</artifactId>
71-
<version>4.3.0-SNAPSHOT</version>
71+
<version>4.3.0</version>
7272
<scope>test</scope>
7373
<exclusions>
7474
<exclusion>
@@ -96,7 +96,7 @@
9696
<dependency>
9797
<groupId>com.alibaba.datax</groupId>
9898
<artifactId>datax-common</artifactId>
99-
<version>4.3.0-SNAPSHOT</version>
99+
<version>4.3.0</version>
100100
<scope>provided</scope>
101101
</dependency>
102102
<dependency>

‎tis-datax/tis-aliyun-jindo-sdk-extends/tis-aliyun-jindo-sdk-extends-impl/dependency-reduced-pom.xml‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,12 @@
33
<parent>
44
<artifactId>tis-aliyun-jindo-sdk-extends</artifactId>
55
<groupId>com.qlangtech.tis.plugins</groupId>
6-
<version>4.3.0-SNAPSHOT</version>
6+
<version>4.3.0</version>
77
</parent>
88
<modelVersion>4.0.0</modelVersion>
99
<groupId>com.qlangtech.tis.plugins</groupId>
1010
<artifactId>tis-aliyun-jindo-sdk-extends-impl</artifactId>
11-
<version>4.3.0-SNAPSHOT</version>
11+
<version>4.3.0</version>
1212
<licenses>
1313
<license>
1414
<name>GNU Affero General Public License</name>
@@ -46,7 +46,7 @@
4646
<dependency>
4747
<groupId>com.alibaba.datax</groupId>
4848
<artifactId>datax-common</artifactId>
49-
<version>4.3.0-SNAPSHOT</version>
49+
<version>4.3.0</version>
5050
<scope>provided</scope>
5151
</dependency>
5252
<dependency>
@@ -64,7 +64,7 @@
6464
<dependency>
6565
<groupId>com.qlangtech.tis</groupId>
6666
<artifactId>tis-plugin</artifactId>
67-
<version>4.3.0-SNAPSHOT</version>
67+
<version>4.3.0</version>
6868
<scope>provided</scope>
6969
</dependency>
7070
<dependency>

‎tis-incr/tis-chunjun-base-plugin/src/main/java/com/qlangtech/tis/plugins/incr/flink/chunjun/script/ChunjunSqlType.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,13 +24,19 @@
2424
import com.qlangtech.tis.extension.TISExtension;
2525
import com.qlangtech.tis.extension.impl.IOUtils;
2626
import com.qlangtech.tis.plugin.IEndTypeGetter;
27+
import com.qlangtech.tis.plugin.IEndTypeGetter.EndType;
2728
import com.qlangtech.tis.plugin.ds.IColMetaGetter;
2829
import com.qlangtech.tis.plugins.incr.flink.connector.ChunjunSinkFactory;
2930
import com.qlangtech.tis.plugins.incr.flink.connector.streamscript.BasicFlinkStreamScriptCreator;
3031
import com.qlangtech.tis.sql.parser.tuple.creator.AdapterStreamTemplateData;
3132
import com.qlangtech.tis.sql.parser.tuple.creator.IStreamIncrGenerateStrategy;
33+
import org.apache.commons.lang.StringUtils;
34+
import org.apache.flink.table.factories.Factory;
3235

36+
import java.util.Iterator;
3337
import java.util.List;
38+
import java.util.ServiceLoader;
39+
import java.util.function.Consumer;
3440

3541
/**
3642
* @author: 百岁（baisui@qlangtech.com）
@@ -42,6 +48,24 @@ public static String getTableSinkTypeName(IEndTypeGetter.EndType endType) {
4248
return "tis-" + endType.getVal() + "-x";
4349
}
4450

51+
@Override
52+
public boolean preValidate(EndType endType, ChunjunSinkFactory sinkFactory, Consumer<String> errorMsgConsumer) {
53+
String tableSinkTypeName = getTableSinkTypeName(endType);
54+
// 校验是否已经定义了对应table DynamicTableSinkFactory
55+
ServiceLoader<Factory> factoriesLoader = ServiceLoader.load(Factory.class, sinkFactory.getClass().getClassLoader());
56+
57+
Iterator<Factory> iterator = factoriesLoader.iterator();
58+
Factory factory = null;
59+
while (iterator.hasNext()) {
60+
factory = iterator.next();
61+
if (StringUtils.equalsIgnoreCase(tableSinkTypeName, factory.factoryIdentifier())) {
62+
return true;
63+
}
64+
}
65+
errorMsgConsumer.accept("未定义对应的Table Sink:'" + tableSinkTypeName + "'，请联系系统管理员");
66+
return false;
67+
}
68+
4569
@Override
4670
public BasicFlinkStreamScriptCreator createStreamTableCreator(
4771
IStreamTableMeataCreator.ISinkStreamMetaCreator sinkStreamMetaCreator) {

‎tis-incr/tis-chunjun-base-plugin/src/main/java/com/qlangtech/tis/plugins/incr/flink/chunjun/script/ChunjunStreamScriptType.java‎

Lines changed: 29 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1,30 +1,44 @@
11
/**
2-
* Licensed to the Apache Software Foundation (ASF) under one
3-
* or more contributor license agreements. See the NOTICE file
4-
* distributed with this work for additional information
5-
* regarding copyright ownership. The ASF licenses this file
6-
* to you under the Apache License, Version 2.0 (the
7-
* "License"); you may not use this file except in compliance
8-
* with the License. You may obtain a copy of the License at
9-
*
10-
* http://www.apache.org/licenses/LICENSE-2.0
11-
*
12-
* Unless required by applicable law or agreed to in writing, software
13-
* distributed under the License is distributed on an "AS IS" BASIS,
14-
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15-
* See the License for the specific language governing permissions and
16-
* limitations under the License.
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
* <p>
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
* <p>
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
1717
*/
1818

1919
package com.qlangtech.tis.plugins.incr.flink.chunjun.script;
2020

2121
import com.qlangtech.tis.extension.Describable;
22+
import com.qlangtech.tis.plugin.IEndTypeGetter.EndType;
23+
import com.qlangtech.tis.plugins.incr.flink.connector.ChunjunSinkFactory;
2224
import com.qlangtech.tis.plugins.incr.flink.connector.scripttype.IStreamScriptType;
25+
import com.qlangtech.tis.runtime.module.misc.IFieldErrorHandler;
26+
27+
import java.util.function.Consumer;
2328

2429
/**
2530
* @author: 百岁（baisui@qlangtech.com）
2631
* @create: 2022-11-21 10:32
2732
**/
2833
public abstract class ChunjunStreamScriptType implements Describable<ChunjunStreamScriptType>, IStreamScriptType {
2934

35+
/**
36+
* 前置校验ScriptType是否正确
37+
*
38+
* @param errorMsgConsumer
39+
* @return
40+
* @see com.qlangtech.tis.plugins.incr.flink.connector.ChunjunSinkFactory.BasicChunjunSinkDescriptor validateAll(IFieldErrorHandler, com.qlangtech.tis.runtime.module.misc.Context) 中被调用
41+
*/
42+
public abstract boolean preValidate(EndType endType, ChunjunSinkFactory sinkFactory, Consumer<String> errorMsgConsumer);
43+
3044
}

‎tis-incr/tis-chunjun-base-plugin/src/main/java/com/qlangtech/tis/plugins/incr/flink/chunjun/script/StreamApiScript.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,12 @@
2121
import com.qlangtech.tis.datax.IStreamTableMeataCreator;
2222
import com.qlangtech.tis.extension.Descriptor;
2323
import com.qlangtech.tis.extension.TISExtension;
24+
import com.qlangtech.tis.plugin.IEndTypeGetter.EndType;
25+
import com.qlangtech.tis.plugins.incr.flink.connector.ChunjunSinkFactory;
2426
import com.qlangtech.tis.plugins.incr.flink.connector.streamscript.BasicFlinkStreamScriptCreator;
2527

28+
import java.util.function.Consumer;
29+
2630
/**
2731
* @author: 百岁（baisui@qlangtech.com）
2832
* @create: 2022-11-14 12:40
@@ -34,6 +38,11 @@ public BasicFlinkStreamScriptCreator createStreamTableCreator(
3438
return new StreamAPIStreamScriptCreator(sinkStreamMetaCreator);
3539
}
3640

41+
@Override
42+
public boolean preValidate(EndType endType, ChunjunSinkFactory sinkFactory, Consumer<String> errorMsgConsumer) {
43+
return true;
44+
}
45+
3746
static class StreamAPIStreamScriptCreator extends BasicFlinkStreamScriptCreator {
3847

3948
public StreamAPIStreamScriptCreator(IStreamTableMeataCreator.ISinkStreamMetaCreator sinkStreamMetaGetter) {

‎tis-incr/tis-chunjun-base-plugin/src/main/java/com/qlangtech/tis/plugins/incr/flink/connector/ChunjunSinkFactory.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@
5454
import com.qlangtech.tis.datax.TableAlias;
5555
import com.qlangtech.tis.datax.TableAliasMapper;
5656
import com.qlangtech.tis.datax.impl.DataxWriter;
57+
import com.qlangtech.tis.extension.Describable;
5758
import com.qlangtech.tis.extension.Descriptor;
5859
import com.qlangtech.tis.manage.common.Option;
5960
import com.qlangtech.tis.plugin.annotation.FormField;
@@ -105,6 +106,8 @@
105106
import java.util.concurrent.ExecutionException;
106107
import java.util.concurrent.TimeUnit;
107108
import java.util.concurrent.atomic.AtomicReference;
109+
import java.util.function.Consumer;
110+
import java.util.function.Function;
108111
import java.util.function.Supplier;
109112
import java.util.stream.Collectors;
110113

@@ -679,7 +682,9 @@ public final String getDisplayName() {
679682
@Override
680683
protected boolean validateAll(IControlMsgHandler msgHandler
681684
, Context context, PostFormVals postFormVals) {
682-
return super.validateAll(msgHandler, context, postFormVals);
685+
ChunjunSinkFactory sinkFactory = postFormVals.newInstance();
686+
Consumer<String> errorMsgConsumer = (errorMsg) -> msgHandler.addFieldError(context, "scriptType", errorMsg);
687+
return sinkFactory.scriptType.preValidate(this.getChunjunEndType(), sinkFactory, errorMsgConsumer);
683688
}
684689

685690
@Override

‎tis-incr/tis-flink-cdc-common/src/main/java/com/qlangtech/plugins/incr/debuzium/DebuziumPropAssist.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,8 +46,8 @@ public static <T extends Describable, PLUGIN extends T> Options<PLUGIN, Field> c
4646
}
4747

4848
@Override
49-
protected String getDescription(Field configOption) {
50-
return configOption.description();
49+
protected MarkdownHelperContent getDescription(Field configOption) {
50+
return new MarkdownHelperContent(configOption.description());
5151
}
5252

5353
@Override

‎tis-incr/tis-flink-cdc-kingbase-shade-4-debezium-connector-postgresql/dependency-reduced-pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,12 @@
33
<parent>
44
<artifactId>tis-incr</artifactId>
55
<groupId>com.qlangtech.tis.plugins</groupId>
6-
<version>4.3.0-SNAPSHOT</version>
6+
<version>4.3.0</version>
77
</parent>
88
<modelVersion>4.0.0</modelVersion>
99
<groupId>com.qlangtech.tis.plugins</groupId>
1010
<artifactId>tis-flink-cdc-kingbase-shade-4-debezium-connector-postgresql</artifactId>
11-
<version>4.3.0-SNAPSHOT</version>
11+
<version>4.3.0</version>
1212
<licenses>
1313
<license>
1414
<name>GNU Affero General Public License</name>
@@ -69,7 +69,7 @@
6969
<dependency>
7070
<groupId>com.qlangtech.tis</groupId>
7171
<artifactId>tis-plugin</artifactId>
72-
<version>4.3.0-SNAPSHOT</version>
72+
<version>4.3.0</version>
7373
<scope>provided</scope>
7474
</dependency>
7575
<dependency>

‎tis-incr/tis-flink-cdc-mysql-shade-4-debezium-connector-mysql/dependency-reduced-pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,12 @@
33
<parent>
44
<artifactId>tis-incr</artifactId>
55
<groupId>com.qlangtech.tis.plugins</groupId>
6-
<version>4.3.0-SNAPSHOT</version>
6+
<version>4.3.0</version>
77
</parent>
88
<modelVersion>4.0.0</modelVersion>
99
<groupId>com.qlangtech.tis.plugins</groupId>
1010
<artifactId>tis-flink-cdc-shade-4-debezium-connector-mysql</artifactId>
11-
<version>4.3.0-SNAPSHOT</version>
11+
<version>4.3.0</version>
1212
<licenses>
1313
<license>
1414
<name>GNU Affero General Public License</name>
@@ -66,7 +66,7 @@
6666
<dependency>
6767
<groupId>com.qlangtech.tis</groupId>
6868
<artifactId>tis-plugin</artifactId>
69-
<version>4.3.0-SNAPSHOT</version>
69+
<version>4.3.0</version>
7070
<scope>provided</scope>
7171
</dependency>
7272
<dependency>

‎tis-incr/tis-flink-cdc-postgresql-shade-4-debezium-connector-postgresql/dependency-reduced-pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,12 @@
33
<parent>
44
<artifactId>tis-incr</artifactId>
55
<groupId>com.qlangtech.tis.plugins</groupId>
6-
<version>4.3.0-SNAPSHOT</version>
6+
<version>4.3.0</version>
77
</parent>
88
<modelVersion>4.0.0</modelVersion>
99
<groupId>com.qlangtech.tis.plugins</groupId>
1010
<artifactId>tis-flink-cdc-postgresql-shade-4-debezium-connector-postgresql</artifactId>
11-
<version>4.3.0-SNAPSHOT</version>
11+
<version>4.3.0</version>
1212
<licenses>
1313
<license>
1414
<name>GNU Affero General Public License</name>
@@ -68,7 +68,7 @@
6868
<dependency>
6969
<groupId>com.qlangtech.tis</groupId>
7070
<artifactId>tis-plugin</artifactId>
71-
<version>4.3.0-SNAPSHOT</version>
71+
<version>4.3.0</version>
7272
<scope>provided</scope>
7373
</dependency>
7474
<dependency>

0 commit comments

Comments
 (0)