cdc 2.2.1 cdc Incremental snapshot for tables requires primary key, but table doesn't have primary key

[JavaKingCr]

使用 Datastream Api 将 Mysql Binlog 收集到 Kafka 时，我发现当某些表没有主键时会出错
我的代码：
val env = StreamExecutionEnvironment.getExecutionEnvironment
checkPoint(env)
val mysqlCDCSource = MySqlSource.builder().hostname(Common.mysqlHostName).port(3306)
.databaseList(Common.mysqlPBQDataBase)
.tableList("")
.username(Common.mysqlUserName)
.password(Common.mysqlPassword)
.startupOptions(StartupOptions.initial())
// .deserializer(new JsonDebeziumDeserializationSchema)
// .deserializer(new StringDebeziumDeserializationSchema)
.deserializer(new CustomerDeserialization)
.includeSchemaChanges(true)
.build()

  flink  版本1.13.6 
  cdc 版本 2.2.1

[peter-cd]

同问下，在sql下可以设置 with( 'scan.incremental.snapshot.enabled' = 'false')，但是在datastream api下并没有提供设置的方法啊，这导致无主键表无法同步啊

[JavaKingCr]

源码：
public static Column getChunkKeyColumn(Table table, https://github.com/nullable String chunkKeyColumn) {
List primaryKeys = table.primaryKeyColumns();

if (primaryKeys.isEmpty()) {
    throw new ValidationException(
            String.format(
                    "Incremental snapshot for tables requires primary key,"
                            + " but table %s doesn't have primary key.",
                    table.id()));

这个地方没有主键直接抛出异常了