Merge pull request #748 from tapdata/develop

Develop

Author: jarad0628

.github/workflows/mr-ci.yaml

@@ -44,7 +44,7 @@ jobs:
       TAG_NAME: ${{ steps.set-output.outputs.TAG_NAME }}
     steps:
       - name: Checkout Tapdata Opensource
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           repository: 'tapdata/tapdata'
           token: ${{ secrets.TAPDATA_ENT_CICD_TOKEN }}
@@ -73,33 +73,12 @@ jobs:
           echo "::set-output name=TAG_NAME::${TAG_NAME}"
 
   Sync-Code-to-Office:
-    runs-on: ubuntu-latest
-    needs:
-      - Get-Stable-Branch
-    steps:
-      - name: Checkout Tapdata Connectors Code
-        uses: actions/checkout@v3
-        with:
-          repository: 'tapdata/tapdata-connectors'
-          ref: ${{ needs.Get-Stable-Branch.outputs.CONNECTORS_BRANCH }}
-          token: ${{ secrets.TAPDATA_ENT_CICD_TOKEN }}
-          path: tapdata-connectors
-          fetch-depth: 0
-      - name: Checkout Tapdata Application
-        uses: actions/checkout@v2
-        with:
-          repository: 'tapdata/tapdata-application'
-          ref: main
-          token: ${{ secrets.TAPDATA_ENT_CICD_TOKEN }}
-          path: tapdata-application
-          fetch-depth: 0
-      - name: Push Tapdata Connectors to Gogs
-        uses: nick-fields/retry@v2
-        with:
-          timeout_minutes: 10
-          max_attempts: 3
-          command: |
-            bash tapdata-application/build/upgrade.sh --upgrade-code=true --upgrade-code-path=tapdata-connectors --gogs-project-name=tapdata-connectors
+    needs: Get-Stable-Branch
+    uses: tapdata/tapdata-application/.github/workflows/sync-code-to-office.yaml@main
+    secrets: inherit
+    with:
+      tapdata-connectors: ${{ needs.Get-Current-Branch.outputs.branch }}
+      gitee-token-user: "${{ vars.GITEE_TOKEN_USER }}"
 
   Scan-Connectors:
     runs-on: ubuntu-latest
@@ -124,7 +103,7 @@ jobs:
           wait_workflow: true
       - name: Checkout Tapdata-Application
         if: ${{ always() && steps.sonar.outcome == 'failure' }}
-        uses: actions/checkout@v2
+        uses: actions/checkout@v4
         with:
           repository: 'tapdata/tapdata-application'
           ref: "main"

connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/PostgresTest.java

@@ -4,11 +4,14 @@
 import io.tapdata.common.CommonDbTest;
 import io.tapdata.common.util.FileUtil;
 import io.tapdata.connector.postgres.config.PostgresConfig;
+import io.tapdata.connector.postgres.error.PostgresErrorCode;
 import io.tapdata.entity.simplify.TapSimplify;
+import io.tapdata.exception.TapCodeException;
 import io.tapdata.kit.EmptyKit;
 import io.tapdata.kit.StringKit;
 import io.tapdata.pdk.apis.entity.ConnectionOptions;
 import io.tapdata.pdk.apis.entity.TestItem;
+import io.tapdata.pdk.apis.exception.TapTestItemException;
 import io.tapdata.pdk.apis.exception.testItem.TapTestCurrentTimeConsistentEx;
 import io.tapdata.pdk.apis.exception.testItem.TapTestReadPrivilegeEx;
 import io.tapdata.pdk.apis.exception.testItem.TapTestStreamReadEx;
@@ -139,19 +142,24 @@ public Boolean testStreamRead() {
         properties.put("password", commonDbConfig.getPassword());
         properties.put("replication", "database");
         properties.put("assumeMinServerVersion", "9.4");
+        List<String> testSqls = TapSimplify.list();
+        long begin = System.currentTimeMillis();
         try {
             Connection connection = new Driver().connect(commonDbConfig.getDatabaseUrl(), properties);
             assert connection != null;
             connection.close();
-            List<String> testSqls = TapSimplify.list();
             String testSlotName = "test_tapdata_" + UUID.randomUUID().toString().replaceAll("-", "_");
             testSqls.add(String.format(PG_LOG_PLUGIN_CREATE_TEST, testSlotName, ((PostgresConfig) commonDbConfig).getLogPluginName()));
             testSqls.add(PG_LOG_PLUGIN_DROP_TEST);
-            jdbcContext.batchExecute(testSqls);
+            jdbcContext.batchExecute(testSqls, 20);
             consumer.accept(testItem(TestItem.ITEM_READ_LOG, TestItem.RESULT_SUCCESSFULLY, "Cdc can work normally"));
             return true;
         } catch (Throwable e) {
-            consumer.accept(new TestItem(TestItem.ITEM_READ_LOG, new TapTestStreamReadEx(e), TestItem.RESULT_SUCCESSFULLY_WITH_WARN));
+            if (System.currentTimeMillis() - begin > 18000) {
+                consumer.accept(new TestItem(TestItem.ITEM_READ_LOG, new TapTestItemException(new TapCodeException(PostgresErrorCode.CREATE_SLOT_TIMEOUT).dynamicDescriptionParameters(String.join(",", testSqls))), TestItem.RESULT_SUCCESSFULLY_WITH_WARN));
+            } else {
+                consumer.accept(new TestItem(TestItem.ITEM_READ_LOG, new TapTestItemException(new TapCodeException(PostgresErrorCode.CREATE_SLOT_FAILED).dynamicDescriptionParameters(String.join(",", testSqls))), TestItem.RESULT_SUCCESSFULLY_WITH_WARN));
+            }
             return null;
         }
     }

connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/PostgresCdcRunner.java

@@ -8,7 +8,6 @@
 import io.tapdata.connector.postgres.PostgresJdbcContext;
 import io.tapdata.connector.postgres.cdc.config.PostgresDebeziumConfig;
 import io.tapdata.connector.postgres.cdc.offset.PostgresOffset;
-import io.tapdata.connector.postgres.cdc.offset.PostgresOffsetStorage;
 import io.tapdata.connector.postgres.config.PostgresConfig;
 import io.tapdata.entity.event.TapEvent;
 import io.tapdata.entity.event.control.HeartbeatEvent;

connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/cdc/offset/PostgresOffsetStorage.java

@@ -54,4 +54,58 @@ public interface PostgresErrorCode {
             type = TapExType.RUNTIME
     )
     String SELECT_PUBLICATION_FAILED = "410002";
+
+    @TapExCode(
+            describe = "Failed to create slot, please analyze the specific error message. \n" +
+                    "Common errors and reasons: \n" +
+                    "ERROR: must be superuser or replication role to use replication slots. \n" +
+                    "The current user does not have permission to create slots.",
+            describeCN = "创建 slot 失败，需结合具体报错信息分析。\n" +
+                    "常见的报错及原因：\n" +
+                    "ERROR: must be superuser or replication role to use replication slots \n" +
+                    "当前用户没有创建 slot 的权限。",
+            solution = "Solution (choose one): \n" +
+                    "1. Use a superuser connection (such as postgres). \n" +
+                    "2. Grant the current user the required permissions and execute: ALTER USER username REPLICATION;",
+            solutionCN = "解决方案（任选其一）：\n" +
+                    "1. 使用超级用户连接（例如 postgres）。\n" +
+                    "2. 赋予当前用户所需权限，执行：ALTER USER username REPLICATION;",
+            dynamicDescription = "Execute sql failed: {}",
+            dynamicDescriptionCN = "执行语句失败：{}",
+            level = TapExLevel.CRITICAL,
+            type = TapExType.RUNTIME
+    )
+    String CREATE_SLOT_FAILED = "410003";
+
+    @TapExCode(
+            describe = "Failed to create slot, please analyze the specific error message. \n" +
+                    "Common errors and reasons: \n" +
+                    "ERROR: timeout expired. \n" +
+                    "The creation of slot timed out, which is probably affected by a long time of uncommitted transaction or other database processes blocking wait.",
+            describeCN = "创建 slot 超时，大概率是受长时间未提交事务影响或其它数据库进程阻塞等待",
+            solution = "Solution (cautious handling): \n" +
+                    "1. Please check if there are any long uncommitted transactions in the database. If so, please commit them first. \n" +
+                    "2. Please check if there are any other database processes blocking wait. If so, please handle them first.\n" +
+                    "SELECT pid, datname, usename, state, xact_start, now() - xact_start AS duration, query\n" +
+                    "FROM pg_stat_activity\n" +
+                    "WHERE state != 'idle'\n" +
+                    "  AND xact_start IS NOT NULL\n" +
+                    "  AND now() - xact_start > interval '300 seconds'  -- for example, find transactions that have been running for more than 5 minutes\n" +
+                    "ORDER BY xact_start;",
+            solutionCN = "解决方案（需谨慎处理）：\n" +
+                    "1. 请检查数据库是否存在长时间未提交事务，如存在请先提交事务。\n" +
+                    "SELECT pid, datname, usename, state, xact_start, now() - xact_start AS duration, query\n" +
+                    "FROM pg_stat_activity\n" +
+                    "WHERE state != 'idle'\n" +
+                    "  AND xact_start IS NOT NULL\n" +
+                    "  AND now() - xact_start > interval '300 seconds'  -- 例如，查找超过5分钟的事务\n" +
+                    "ORDER BY xact_start;\n" +
+                    "2. 请检查数据库是否存在其它数据库进程阻塞等待，如存在请先处理阻塞。",
+            dynamicDescription = "Execute sql failed: {}",
+            dynamicDescriptionCN = "执行语句失败：{}",
+            level = TapExLevel.CRITICAL,
+            type = TapExType.RUNTIME,
+            recoverable = true
+    )
+    String CREATE_SLOT_TIMEOUT = "410004";
 }

connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/error/PostgresErrorCode.java

@@ -227,6 +227,17 @@ public void execute(String sql) throws SQLException {
         }
     }
 
+    public void execute(String sql, int timeout) throws SQLException {
+        try (
+                Connection connection = getConnection();
+                Statement statement = connection.createStatement()
+        ) {
+            statement.setQueryTimeout(timeout);
+            statement.execute(sql);
+            connection.commit();
+        }
+    }
+
     public void batchExecute(List<String> sqlList) throws SQLException {
         try (
                 Connection connection = getConnection();
@@ -239,6 +250,19 @@ public void batchExecute(List<String> sqlList) throws SQLException {
         }
     }
 
+    public void batchExecute(List<String> sqlList, int timeout) throws SQLException {
+        try (
+                Connection connection = getConnection();
+                Statement statement = connection.createStatement()
+        ) {
+            statement.setQueryTimeout(timeout);
+            for (String sql : sqlList) {
+                statement.execute(sql);
+            }
+            connection.commit();
+        }
+    }
+
     public void queryAllTables(List<String> tableNames, int batchSize, Consumer<List<String>> consumer) throws SQLException {
         List<String> temp = list();
         query(queryAllTablesSql(getConfig().getSchema(), tableNames),

connectors-common/sql-core/src/main/java/io/tapdata/common/JdbcContext.java

@@ -24,7 +24,7 @@ public class DorisConfig extends CommonDbConfig {
     private List<LinkedHashMap<String, String>> tableProperties = new ArrayList<>();
     private Boolean jdbcCompletion = false;
 
-    private Boolean useHTTPS =false;
+    private Boolean useHTTPS = false;
 
     private Integer backendNum;
 
@@ -38,9 +38,9 @@ public DorisConfig() {
     public DorisConfig load(Map<String, Object> map) {
         DorisConfig config = (DorisConfig) super.load(map);
         config.setSchema(config.getDatabase());
-        if(Boolean.TRUE.equals(useHTTPS)){
+        if (Boolean.TRUE.equals(useHTTPS)) {
             config.setDorisHttp(getDorisHttp().replace("https://", ""));
-        }else{
+        } else {
             config.setDorisHttp(getDorisHttp().replace("http://", ""));
         }
         return config;
@@ -109,10 +109,6 @@ public String getWriteFormat() {
         return writeFormat;
     }
 
-    public String getWriteFormat(String key) {
-        return getTableConfigValue(key, "writeFormat", writeFormat);
-    }
-
     public void setWriteFormat(String writeFormat) {
         this.writeFormat = writeFormat;
     }
@@ -185,8 +181,4 @@ public enum WriteFormat {
     public WriteFormat getWriteFormatEnum() {
         return WriteFormat.valueOf(writeFormat);
     }
-
-    public WriteFormat getWriteFormatEnum(String key) {
-        return WriteFormat.valueOf(getTableConfigValue(key, "writeFormat", writeFormat));
-    }
 }

connectors/doris-connector/src/main/java/io/tapdata/connector/doris/bean/DorisConfig.java

@@ -71,7 +71,7 @@ public DorisStreamLoader(DorisJdbcContext dorisJdbcContext, CloseableHttpClient
     }
 
     private void initMessageSerializer() {
-        DorisConfig.WriteFormat writeFormat = dorisConfig.getWriteFormatEnum(tapTable.getId());
+        DorisConfig.WriteFormat writeFormat = dorisConfig.getWriteFormatEnum();
         TapLogger.info(TAG, "Doris stream load run with {} format", writeFormat);
         switch (writeFormat) {
             case csv:
@@ -137,7 +137,7 @@ private Set<String> getDataColumns(TapRecordEvent recordEvent) {
     }
 
     public RespContent put(final TapTable table) throws StreamLoadException, DorisRetryableException {
-        DorisConfig.WriteFormat writeFormat = dorisConfig.getWriteFormatEnum(table.getId());
+        DorisConfig.WriteFormat writeFormat = dorisConfig.getWriteFormatEnum();
         try {
             final String loadUrl = buildLoadUrl(dorisConfig.getDorisHttp(), dorisConfig.getDatabase(), table.getId());
             final String prefix = buildPrefix(URLEncoder.encode(table.getId()).replace("%", ""));

connectors/doris-connector/src/main/java/io/tapdata/connector/doris/streamload/DorisStreamLoader.java

@@ -487,7 +487,6 @@
           "title": "${writeFormat}",
           "default": "json",
           "x-index": 7,
-          "x-perTable": true,
           "x-decorator": "FormItem",
           "x-decorator-props": {
             "feedbackLayout": "none"

connectors/doris-connector/src/main/resources/spec_doris.json

@@ -1,12 +1,6 @@
 package io.tapdata.connector.config;
 
 import io.tapdata.connector.IConfigWithContext;
-import io.tapdata.connector.tester.IStep;
-import io.tapdata.connector.tester.items.ClusterURITesterItem;
-import io.tapdata.pdk.apis.entity.ConnectionOptions;
-import io.tapdata.pdk.apis.entity.TestItem;
-
-import java.util.function.Consumer;
 
 /**
  * 测试项-集群地址
@@ -21,15 +15,4 @@ default String getConnectionClusterURI() {
         return connectionConfigGet(KEY_CLUSTER_URI, null);
     }
 
-    default boolean testClusterURI(TestItem item, Consumer<TestItem> consumer, ConnectionOptions options) {
-        String clusterURI = getConnectionClusterURI();
-        if (null == clusterURI) {
-            item.setResult(TestItem.RESULT_FAILED);
-            item.setInformation(String.format("not configured '%s' yet", KEY_CLUSTER_URI));
-        } else {
-            options.connectionString(clusterURI);
-            ClusterURITesterItem.test(clusterURI, item, consumer);
-        }
-        return IStep.CHECK_ITEM_APPLY;
-    }
 }