如何解决SQL Server间基于Flink CDC的数据同步报错问题？

环境信息 #

• Flink 1.19.1
• Flink CDC 3.1.0
• SQL Server 2022
• OpenJDK 11.0.23

已执行步骤 #

1. 参照Flink官方文档编写示例代码；
2. 在IntelliJ IDEA中创建Java应用；
3. 运行App.java时出现错误；
4. 若步骤3成功，则通过Jar包提交任务。

App.java代码 #

package org.example;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import com.ververica.cdc.connectors.sqlserver.SqlServerSource;

public class App {

    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        TableEnvironment tEnv = TableEnvironment.create(settings);
        tEnv.executeSql("CREATE TABLE orders (\n" +
                "id INTEGER  NOT NULL  ,\n" +
                "order_date DATE NOT NULL,\n" +
                "purchaser INTEGER NOT NULL,\n" +
                "quantity INTEGER NOT NULL,\n" +
                "product_id INTEGER NOT NULL\n" +
                ") WITH (\n" +
                "'connector' = 'sqlserver-cdc',  \n" +
                "'hostname' = 'x.x.x.x', \n" +
                "'port' = '1433',              \n" +
                "'username' = 'sa',             \n" +
                "'password' = 'xxxxx',      \n" +
                "'database-name' = 'TEST_FOR_FLINK',  \n" +
                "'table-name' = 'dbo.orders'        \n" +
                ");");
        tEnv.executeSql("\n" +
                "CREATE TABLE orders_dw (\n" +
                "    id INTEGER NOT NULL,\n" +
                "    order_date DATE NOT NULL,\n" +
                "    purchaser INTEGER NOT NULL,\n" +
                "    quantity INTEGER NOT NULL,\n" +
                "product_id INTEGER NOT NULL,\n" +
                "PRIMARY KEY (id) NOT ENFORCED \n" +
                ") WITH (\n" +
                "    'connector' = 'jdbc',                          \n" +
                "    'url' = 'jdbc:sqlserver://x.x.x.x:x;databaseName=DW', \n" +
                "    'username' = 'sa',                  \n" +
                "    'password' = 'xxxxx',                  \n" +
                "    'table-name' = 'dbo.orders_from_flink'                   \n" +
                ");");
        Table transactions = tEnv.from("orders");
        transactions.executeInsert("orders_dw");
    }
}

POM依赖配置 #

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>org.example</groupId>
  <artifactId>FLinkSQL</artifactId>
  <version>1.0-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>FLinkSQL</name>
  <url>http://maven.apache.org</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>11</maven.compiler.source>
    <maven.compiler.target>11</maven.compiler.target>
    <flink.version>1.19.1</flink.version>
    <flink.cdc.version>3.1.0</flink.cdc.version>
    <sqlserver.jdbc.version>12.6.3.jre11</sqlserver.jdbc.version>
  </properties>

  <dependencies>
    <!-- Flink dependencies -->
    <!-- https://mvnrepository.com/artifact/com.ververica/flink-sql-connector-sqlserver-cdc -->
    <dependency>
      <groupId>com.ververica</groupId>
      <artifactId>flink-sql-connector-sqlserver-cdc</artifactId>
      <version>3.0.1</version>
<!--      <scope>provided</scope>-->
    </dependency>

    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-api-java-bridge</artifactId>
      <version>1.19.1</version>
<!--      <scope>provided</scope>-->
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-planner_2.12</artifactId>
      <version>1.19.1</version>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>com.microsoft.sqlserver</groupId>
      <artifactId>mssql-jdbc</artifactId>
      <version>12.6.3.jre11</version>
    </dependency>

    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-test-utils</artifactId>
      <version>1.19.1</version>
      <scope>test</scope>
    </dependency>

  </dependencies>

  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>3.8.1</version>
        <configuration>
          <source>${maven.compiler.source}</source>
          <target>${maven.compiler.target}</target>
        </configuration>
      </plugin>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-jar-plugin</artifactId>
        <version>3.2.0</version>
        <configuration>
          <archive>
            <manifest>
              <addClasspath>true</addClasspath>
              <mainClass>org.example.App</mainClass>
            </manifest>
          </archive>
        </configuration>
      </plugin>
    </plugins>
  </build>
</project>

错误信息 #

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Exception in thread "main" org.apache.flink.table.api.ValidationException: Could not find any factories that implement 'org.apache.flink.table.delegation.ExecutorFactory' in the classpath.
    at org.apache.flink.table.factories.FactoryUtil.discoverFactory(FactoryUtil.java:605)
    at org.apache.flink.table.api.internal.TableEnvironmentImpl.create(TableEnvironmentImpl.java:241)
    at org.apache.flink.table.api.TableEnvironment.create(TableEnvironment.java:99)
    at org.example.App.main(App.java:12)

运行main方法时出现上述错误，但测试环境中使用Flink SQL CLI可正常运行，恳请帮忙解决该问题，若需更多信息请告知。

另外想了解如何分析和解决此类问题，比如怀疑和依赖包有关，能否推荐相关关键词或文档供学习？

内容的提问来源于stack exchange，提问作者潘德拉贡阿尔托莉雅