add Http-Proxy for TIS

Author: baisui1981

tis-datax/tis-datax-common-plugin/src/main/java/com/qlangtech/tis/plugin/aliyun/AliyunEndpoint.java

@@ -7,6 +7,7 @@
 import com.qlangtech.tis.plugin.AuthToken.IAliyunAccessKey;
 import com.qlangtech.tis.plugin.HttpEndpoint;
 import com.qlangtech.tis.plugin.HttpEndpoint.IAliyunEndpoint;
+import com.qlangtech.tis.plugin.IEndTypeGetter;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 
@@ -43,11 +44,16 @@ public String getEndpointHost() {
     }
 
     @TISExtension()
-    public static class DefaultDescriptor extends HttpEndpoint.DefaultDescriptor {
+    public static class DefaultDescriptor extends HttpEndpoint.DefaultDescriptor implements IEndTypeGetter {
         public DefaultDescriptor() {
             super(KEY_FIELD_ALIYUN_TOKEN);
         }
 
+        @Override
+        public EndType getEndType() {
+            return EndType.Aliyun;
+        }
+
         @Override
         public String getDisplayName() {
             return KEY_FIELD_ALIYUN_TOKEN;

tis-datax/tis-datax-common-plugin/src/main/java/com/qlangtech/tis/plugin/datax/DataFlowDataXProcessor.java

@@ -116,10 +116,6 @@ public IDataxReader getReader(IPluginContext pluginCtx) {
         throw new UnsupportedOperationException("dataflow processor not support single reader getter");
     }
 
-    @Override
-    public TableAliasMapper getTabAlias(IPluginContext pluginCtx) {
-        return TableAliasMapper.Null;
-    }
 
     @Override
     public void afterSaved(IPluginContext pluginContext, Optional<Context> context) {
@@ -419,6 +415,11 @@ public DataXCfgGenerator.GenerateCfgs getDataxCfgFileNames(IPluginContext plugin
         return DataxProcessor.getDataxCfgFileNames(pluginCtx, partialTrigger, this);
     }
 
+    @Override
+    public TableAliasMapper getTabAlias(IPluginContext pluginCtx, boolean withDft) {
+        return TableAliasMapper.Null;
+    }
+
 
     @TISExtension()
     public static class DescriptorImpl extends Descriptor<IAppSource> implements IEndTypeGetter {

tis-datax/tis-datax-common-plugin/src/main/java/com/qlangtech/tis/plugin/datax/common/BasicDataXRdbmsWriter.java

@@ -171,7 +171,7 @@ public void run() {
 
             final TableAliasMapper tableAliasMapper
                     = execContext.getAttribute(TableAlias.class.getSimpleName(), () -> {
-                return execContext.getProcessor().getTabAlias(null);
+                return execContext.getProcessor().getTabAlias(null, true);
             });
 
             BasicDataSourceFactory dsFactory = ((BasicDataSourceFactory) getDataSourceFactory());

tis-datax/tis-datax-dfs-plugin/src/main/java/com/qlangtech/tis/plugin/datax/resmatcher/WildcardDFSResMatcher.java

@@ -65,7 +65,7 @@ public static Optional<TableMap> getTableMap(IPluginContext pluginContext, Strin
             throw new IllegalArgumentException("param dataXName can not be empty");
         }
         IDataxProcessor dataxProcessor = DataxProcessor.load(pluginContext, dataXName);
-        TableAliasMapper tabAlias = dataxProcessor.getTabAlias(pluginContext);
+        TableAliasMapper tabAlias = dataxProcessor.getTabAlias(pluginContext, true);
         Optional<TableMap> tabAlia = tabAlias.getFirstTableMap();
         return tabAlia;
     }
@@ -94,7 +94,7 @@ public List<ColumnMetaData> getTableMetadata(IPluginContext pluginContext, Strin
      */
     @Override
     public SourceColsMeta getSourceColsMeta(ITDFSSession hdfsSession, Optional<String> entityName, String path, IDataxProcessor processor) {
-        TableAliasMapper tabAlias = processor.getTabAlias(null);
+        TableAliasMapper tabAlias = processor.getTabAlias(null, false);
         Optional<TableAlias> findMapper = tabAlias.findFirst();
         IDataxProcessor.TableMap tabMapper
                 = (IDataxProcessor.TableMap) findMapper.orElseThrow(() -> new NullPointerException("TableAlias can not be null"));
@@ -110,7 +110,7 @@ public List<ISelectedTab> getSelectedTabs(IDFSReader dfsReader) {
         }
 
         IDataxProcessor processor = DataxProcessor.load(IPluginContext.getThreadLocalInstance(), reader.dataXName);
-        TableAliasMapper tabAlias = processor.getTabAlias(IPluginContext.getThreadLocalInstance());
+        TableAliasMapper tabAlias = processor.getTabAlias(IPluginContext.getThreadLocalInstance(), false);
         Optional<TableAlias> findMapper = tabAlias.findFirst();
         if (findMapper.isPresent()) {
             IDataxProcessor.TableMap tabMapper = (IDataxProcessor.TableMap) findMapper.get();

tis-datax/tis-datax-dfs-plugin/src/main/resources/com/qlangtech/tis/plugin/datax/DataXDFSReader_introduce.md

@@ -0,0 +1,39 @@
+# DFS数据读取插件使用说明
+
+## 功能概述
+
+本插件用于从分布式文件系统（如HDFS、FTP服务器等）中批量读取数据文件。支持通过匹配规则自动查找文件，无需手工指定每个文件路径。
+
+## 主要配置项
+
+### 1. 文件系统连接（dfsLinker）
+配置目标文件系统的连接信息，包括服务器地址、端口、访问凭证等。系统支持HDFS、FTP等多种分布式存储。
+
+### 2. 文件匹配规则（resMatcher）
+设置如何查找目标文件，支持两种方式：
+- **通配符匹配**：使用 `*.csv`、`data_*.txt` 等模式匹配文件名
+- **多层目录遍历**：可设置遍历深度，自动搜索子目录下的文件
+
+### 3. 文件格式（fileFormat）
+选择数据文件的格式类型（如CSV、文本文件等），并配置分隔符、编码、压缩格式等参数。
+
+## 配置验证
+
+保存配置时，系统会立即连接到文件系统进行验证：
+- 显示找到的文件数量
+- 列出前5个匹配文件的完整路径
+- 如果找不到文件，会提示具体的错误原因
+
+## 适用场景
+
+- 从Hadoop集群批量导入日志文件
+- 定期从FTP服务器读取业务数据文件
+- 处理按日期分目录存储的历史数据
+- 一次性读取多个结构相同的数据文件
+
+## 使用提示
+
+1. 建议先用小范围路径测试匹配规则是否正确
+2. 确认文件格式配置与实际文件一致
+3. 注意文件系统的访问权限设置
+4. 多层目录遍历时，控制好遍历深度以避免扫描过多文件

tis-datax/tis-datax-dolphinscheduler-plugin/src/main/java/com/qlangtech/tis/plugin/datax/doplinscheduler/export/DolphinSchedulerURLBuilder.java

@@ -28,11 +28,9 @@
 import com.qlangtech.tis.manage.common.HttpUtils.PostParam;
 import com.qlangtech.tis.manage.common.PostFormStreamProcess;
 import com.qlangtech.tis.manage.common.TisUTF8;
-import com.qlangtech.tis.plugin.datax.IWorkflowNode;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang.StringUtils;
-import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -156,11 +154,13 @@ public DolphinSchedulerResponse applyPost(List<PostParam> params, Optional<Strea
 
     private DolphinSchedulerResponse applyRequest(List<PostParam> params
             , Optional<StreamErrorProcess> streamErrorProcess, URL applyUrl, HTTPMethod httpMethod) {
-        return HttpUtils.process(applyUrl, params, new PostFormStreamProcess<DolphinSchedulerResponse>() {
-            @Override
-            public List<Header> getHeaders() {
-                return endpoint.appendToken(super.getHeaders());
-            }
+
+        return HttpUtils.process(applyUrl, params, new PostFormStreamProcess<DolphinSchedulerResponse>( //
+                endpoint.appendToken(Lists.newArrayList())) {
+//            @Override
+//            public List<Header> getHeaders() {
+//                return endpoint.appendToken(super.getHeaders());
+//            }
 
             @Override
             public void error(int status, InputStream errstream, IOException e) throws Exception {

tis-datax/tis-datax-dolphinscheduler-plugin/src/main/java/com/qlangtech/tis/plugin/datax/doplinscheduler/export/ExportTISPipelineToDolphinscheduler.java

@@ -210,7 +210,7 @@ protected void afterManipuldateProcess(IPluginContext pluginContext, Optional<Co
         addProjectParameters();
 
 
-        IDataxProcessor dataxProcessor = DataxProcessor.load(pluginContext, itemsProcessor.getOriginIdentityId());
+        IDataxProcessor dataxProcessor = DataxProcessor.load(pluginContext, itemsProcessor.getOriginIdentityId().orElseThrow());
         DSWorkflowPayload workflowPayload = new DSWorkflowPayload(this, dataxProcessor
                 , BasicDistributedSPIDataXJobSubmit.getCommonDAOContext(pluginContext), new DolphinschedulerDistributedSPIDataXJobSubmit());
         WorkflowSPIInitializer<DSWorkflowInstance> workflowSPIInitializer = new WorkflowSPIInitializer<>(workflowPayload);

tis-datax/tis-datax-elasticsearch-plugin/src/main/java/com/qlangtech/tis/plugin/datax/DataXElasticsearchWriter.java

@@ -244,7 +244,7 @@ public Void varcharType(DataType type) {
     @Override
     public List<ESColumn> initialIndex(IDataxProcessor dataxProcessor) {
         ESTableAlias esSchema = null;
-        Optional<TableAlias> first = dataxProcessor.getTabAlias(null).findFirst();
+        Optional<TableAlias> first = dataxProcessor.getTabAlias(null, true).findFirst();
         if (first.isPresent()) {
             TableAlias value = first.get();
             if (!(value instanceof ESTableAlias)) {

tis-datax/tis-datax-kafka-plugin/src/main/java/com/qlangtech/tis/plugins/datax/kafka/writer/DataXKafkaWriter.java

@@ -25,10 +25,7 @@
 import com.qlangtech.tis.datax.IDataxProcessor;
 import com.qlangtech.tis.datax.impl.DataxWriter;
 import com.qlangtech.tis.extension.Descriptor;
-import com.qlangtech.tis.extension.ElementPluginDesc;
 import com.qlangtech.tis.extension.TISExtension;
-import com.qlangtech.tis.extension.impl.PropertyType;
-import com.qlangtech.tis.extension.impl.SuFormProperties;
 import com.qlangtech.tis.lang.TisException;
 import com.qlangtech.tis.manage.common.TisUTF8;
 import com.qlangtech.tis.plugin.ValidatorCommons;
@@ -39,7 +36,9 @@
 import com.qlangtech.tis.plugin.datax.transformer.RecordTransformerRules;
 import com.qlangtech.tis.plugins.datax.kafka.writer.protocol.KafkaProtocol;
 import com.qlangtech.tis.realtime.transfer.DTO;
+import com.qlangtech.tis.realtime.utils.NetUtils;
 import com.qlangtech.tis.runtime.module.misc.IControlMsgHandler;
+import com.qlangtech.tis.runtime.module.misc.IFieldErrorHandler;
 import com.qlangtech.tis.trigger.util.JsonUtil;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang3.exception.ExceptionUtils;
@@ -51,6 +50,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.net.UnknownHostException;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -150,7 +150,7 @@ public String getTemplate() {
 
     @Override
     public IDataxContext getSubTask(
-            Optional<IDataxProcessor.TableMap> tableMap,Optional<RecordTransformerRules> transformerRules) {
+            Optional<IDataxProcessor.TableMap> tableMap, Optional<RecordTransformerRules> transformerRules) {
         return null;
     }
 
@@ -196,12 +196,44 @@ public Map<String, Object> buildKafkaConfig(boolean isTest) {
     @TISExtension
     public static class DefaultDescriptor extends BaseDataxWriterDescriptor implements DataxWriter.IRewriteSuFormProperties {
 
-       // private transient SuFormProperties rewriteSubFormProperties;
+        // private transient SuFormProperties rewriteSubFormProperties;
 
         public DefaultDescriptor() {
             super();
         }
 
+        //  @Override
+        public boolean validateBootstrapServers(IFieldErrorHandler msgHandler, Context context, String fieldName, String value) {
+
+            String[] servs = org.apache.commons.lang3.StringUtils.split(value, ",");
+            if (servs.length < 1) {
+                msgHandler.addFieldError(context, fieldName, ValidatorCommons.MSG_EMPTY_INPUT_ERROR);
+                return false;
+            }
+
+            for (String host : servs) {
+                if (!Validator.host.validate(msgHandler, context, fieldName, host)) {
+                    return false;
+                }
+                String[] split = org.apache.commons.lang3.StringUtils.split(host, ":");
+                try {
+                    if (!NetUtils.isPortAvailable(split[0], Integer.parseInt(split[1]))) {
+                        msgHandler.addFieldError(context, fieldName, "地址：" + host + " 不可触达，请检查是否可用");
+                        return false;
+                    }
+                } catch (Exception e) {
+                    UnknownHostException unknownHostException = null;
+                    if ((unknownHostException = ExceptionUtils.throwableOfType(e, UnknownHostException.class)) != null) {
+                        msgHandler.addFieldError(context, fieldName, "主机地址：‘" + unknownHostException.getMessage() + "’无法识别");
+                        return false;
+                    }
+                    throw new RuntimeException(e);
+                }
+            }
+
+
+            return true;
+        }
 //        @Override
 //        public SuFormProperties overwriteSubPluginFormPropertyTypes(SuFormProperties subformProps) throws Exception {
 //

tis-datax/tis-datax-test-common/src/main/java/com/qlangtech/tis/plugin/common/BasicTemplate.java

@@ -7,12 +7,11 @@
 import com.qlangtech.tis.datax.IDataxReader;
 import com.qlangtech.tis.datax.IDataxReaderContext;
 import com.qlangtech.tis.datax.IDataxWriter;
+import com.qlangtech.tis.datax.StoreResourceType;
 import com.qlangtech.tis.datax.TableAliasMapper;
 import com.qlangtech.tis.datax.impl.DataXCfgGenerator;
-import com.qlangtech.tis.datax.StoreResourceType;
 import com.qlangtech.tis.datax.impl.TransformerInfo;
 import com.qlangtech.tis.plugin.IPluginStore;
-import com.qlangtech.tis.plugin.datax.SelectedTab;
 import com.qlangtech.tis.plugin.datax.transformer.RecordTransformerRules;
 import com.qlangtech.tis.plugin.ds.ISelectedTab;
 import com.qlangtech.tis.plugin.trigger.JobTrigger;
@@ -113,7 +112,7 @@ public DataXCfgGenerator.GenerateCfgs getDataxCfgFileNames(IPluginContext plugin
             }
 
             @Override
-            public TableAliasMapper getTabAlias(IPluginContext pluginCtx) {
+            public TableAliasMapper getTabAlias(IPluginContext pluginCtx, boolean withDft) {
                 return tableAliasMapper;
             }