add rate limit controller setting support for TIS

Author: baisui1981

tis-asyncmsg-rocketmq-plugin/src/main/java/com/qlangtech/async/message/client/consumer/RocketMQConsumerStatus.java

@@ -47,10 +47,9 @@
 import com.qlangtech.tis.extension.impl.IOUtils;
 import com.qlangtech.tis.job.common.JobCommon;
 import com.qlangtech.tis.manage.common.Config;
-import com.qlangtech.tis.manage.common.DagTaskUtils;
+import com.qlangtech.tis.manage.common.TaskSoapUtils;
 import com.qlangtech.tis.offline.DataxUtils;
 import com.qlangtech.tis.order.center.IAppSourcePipelineController;
-import com.qlangtech.tis.datax.StoreResourceType;
 import com.qlangtech.tis.realtime.transfer.TableSingleDataIndexStatus;
 import com.qlangtech.tis.realtime.utils.NetUtils;
 import com.qlangtech.tis.realtime.yarn.rpc.MasterJob;
@@ -283,7 +282,7 @@ public void exec(final JarLoader uberClassLoader, DataXJobInfo jobName, IDataxPr
             TIS.clean(false);
             if (execMode == DataXJobSubmit.InstanceType.DISTRIBUTE) {
                 try {
-                    DagTaskUtils.feedbackAsynTaskStatus(jobArgs.jobId, jobName.jobFileName, success);
+                    TaskSoapUtils.feedbackAsynTaskStatus(jobArgs.jobId, jobName.jobFileName, success);
                 } catch (Throwable e) {
                     logger.warn("notify exec result faild,jobId:" + jobArgs.jobId + ",jobName:" + jobName, e);
                 }

tis-datax/tis-datax-local-executor-utils/src/main/java/com/qlangtech/tis/datax/DataxExecutor.java

@@ -27,13 +27,15 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.qlangtech.plugins.incr.flink.cdc.SourceChannel;
+import com.qlangtech.plugins.incr.flink.launch.TISFlinkCDCStreamFactory;
 import com.qlangtech.tis.async.message.client.consumer.AsyncMsg;
 import com.qlangtech.tis.async.message.client.consumer.IConsumerHandle;
 import com.qlangtech.tis.async.message.client.consumer.IMQListener;
 import com.qlangtech.tis.async.message.client.consumer.MQConsumeException;
 import com.qlangtech.tis.coredefine.module.action.TargetResName;
 import com.qlangtech.tis.datax.DataXJobInfo;
 import com.qlangtech.tis.datax.DataXJobSubmit;
+import com.qlangtech.tis.datax.DataXName;
 import com.qlangtech.tis.datax.IDataxProcessor;
 import com.qlangtech.tis.datax.IDataxReader;
 import com.qlangtech.tis.datax.IStreamTableMeta;
@@ -44,6 +46,7 @@
 import com.qlangtech.tis.plugin.ds.CMeta;
 import com.qlangtech.tis.plugin.ds.ISelectedTab;
 import com.qlangtech.tis.plugin.ds.TableInDB;
+import com.qlangtech.tis.plugin.incr.IncrStreamFactory;
 import com.qlangtech.tis.realtime.ReaderSource;
 import com.qlangtech.tis.realtime.dto.DTOStream;
 import org.apache.flink.api.common.JobExecutionResult;
@@ -65,6 +68,7 @@
 public abstract class ChunjunSourceFunction
         implements IMQListener<List<ReaderSource>> {
     final ChunjunSourceFactory sourceFactory;
+    IncrStreamFactory streamFactory = new TISFlinkCDCStreamFactory();
 
     public ChunjunSourceFunction(ChunjunSourceFactory sourceFactory) {
         this.sourceFactory = sourceFactory;
@@ -94,7 +98,7 @@ private SourceFunction<RowData> createSourceFunction(
 
 
     @Override
-    public AsyncMsg<List<ReaderSource>> start(boolean flinkCDCPipelineEnable, TargetResName name, IDataxReader dataSource
+    public AsyncMsg<List<ReaderSource>> start(IncrStreamFactory streamFactory, boolean flinkCDCPipelineEnable, DataXName names, IDataxReader dataSource
             , List<ISelectedTab> tabs, IDataxProcessor dataXProcessor) throws MQConsumeException {
         Objects.requireNonNull(dataXProcessor, "dataXProcessor can not be null");
         BasicDataXRdbmsReader reader = (BasicDataXRdbmsReader) dataSource;
@@ -118,7 +122,7 @@ public AsyncMsg<List<ReaderSource>> start(boolean flinkCDCPipelineEnable, Target
                 for (String physicsName : physicsNames) {
                     SyncConf conf = createSyncConf(sourceFactory, jdbcUrl, dbName, (SelectedTab) tab, physicsName);
                     SourceFunction<RowData> sourceFunc = createSourceFunction(tab.getName(), conf, sourceFactory, reader);
-                    sourceFuncs.add(ReaderSource.createRowDataSource(
+                    sourceFuncs.add(ReaderSource.createRowDataSource(streamFactory, names,
                             dbHost + ":" + sourceFactory.port + "_" + dbName + "." + physicsName, tab, sourceFunc));
                 }
             }

tis-incr/tis-chunjun-base-plugin/src/main/java/com/qlangtech/tis/plugins/incr/flink/chunjun/source/ChunjunSourceFunction.java

@@ -0,0 +1,34 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 com.qlangtech.plugins.incr.flink.cdc;
+
+import com.qlangtech.tis.plugin.incr.TISRateLimiter;
+import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
+
+/**
+ * @author: 百岁（baisui@qlangtech.com）
+ * @create: 2025-07-06 12:18
+ * @see com.qlangtech.plugins.incr.flink.cdc.impl.NoRateLimiter
+ * @see com.qlangtech.plugins.incr.flink.cdc.impl.PerSecondRateLimiter
+ **/
+public abstract class BasicRateLimiter extends TISRateLimiter {
+
+    @Override
+    public abstract RateLimiterStrategy getStrategy();
+}

tis-incr/tis-flink-cdc-common/src/main/java/com/qlangtech/plugins/incr/flink/cdc/BasicRateLimiter.java

@@ -19,6 +19,7 @@
 package com.qlangtech.plugins.incr.flink.cdc;
 
 import com.qlangtech.tis.realtime.transfer.DTO;
+import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
 import org.apache.kafka.connect.data.Field;
 
 import java.io.Serializable;
@@ -30,6 +31,7 @@
 public class DefaultSourceValConvert implements ISourceValConvert, Serializable {
     @Override
     public Object convert(DTO dto, Field field, Object val) {
+
         return val;
     }
 }

tis-incr/tis-flink-cdc-common/src/main/java/com/qlangtech/plugins/incr/flink/cdc/DefaultSourceValConvert.java

@@ -0,0 +1,56 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 com.qlangtech.plugins.incr.flink.cdc.impl;
+
+import com.qlangtech.plugins.incr.flink.cdc.BasicRateLimiter;
+
+import com.qlangtech.tis.extension.Descriptor;
+import com.qlangtech.tis.extension.TISExtension;
+import com.qlangtech.tis.plugin.incr.TISRateLimiter;
+import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
+
+/**
+ * 不支持运行期动态设置限流值
+ *
+ * @author: 百岁（baisui@qlangtech.com）
+ * @create: 2025-07-06 12:34
+ **/
+public class NoRateLimiter extends BasicRateLimiter {
+    @Override
+    public RateLimiterStrategy getStrategy() {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public boolean supportRateLimiter() {
+        return false;
+    }
+
+    @TISExtension
+    public static class Desc extends Descriptor<TISRateLimiter> {
+        public Desc() {
+            super();
+        }
+
+        @Override
+        public String getDisplayName() {
+            return SWITCH_OFF;
+        }
+    }
+}

tis-incr/tis-flink-cdc-common/src/main/java/com/qlangtech/plugins/incr/flink/cdc/impl/NoRateLimiter.java

@@ -0,0 +1,68 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 com.qlangtech.plugins.incr.flink.cdc.impl;
+
+import com.alibaba.citrus.turbine.Context;
+import com.qlangtech.plugins.incr.flink.cdc.BasicRateLimiter;
+import com.qlangtech.tis.extension.Descriptor;
+import com.qlangtech.tis.extension.TISExtension;
+import com.qlangtech.tis.plugin.incr.TISRateLimiter;
+import com.qlangtech.tis.runtime.module.misc.IFieldErrorHandler;
+import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
+
+/**
+ * @author: 百岁（baisui@qlangtech.com）
+ * @create: 2025-07-06 12:22
+ **/
+public class PerSecondRateLimiter extends BasicRateLimiter {
+
+//    @FormField(ordinal = 0, type = FormFieldType.INT_NUMBER, validate = {Validator.require, Validator.integer})
+//    public Integer recordsPerSecond;
+
+    @Override
+    public RateLimiterStrategy getStrategy() {
+        return new ResettableRateLimitStrategy(Integer.MAX_VALUE);
+    }
+
+    @Override
+    public boolean supportRateLimiter() {
+        return true;
+    }
+
+    @TISExtension
+    public static class Desc extends Descriptor<TISRateLimiter> {
+        public Desc() {
+            super();
+        }
+
+        public boolean validateRecordsPerSecond(IFieldErrorHandler msgHandler, Context context, String fieldName, String val) {
+            Integer limit = Integer.parseInt(val);
+            if (limit < 100) {
+                msgHandler.addFieldError(context, fieldName, "必须大于100");
+                return false;
+            }
+            return true;
+        }
+
+        @Override
+        public String getDisplayName() {
+            return SWITCH_ON;
+        }
+    }
+}