FlinkCDC的探索与实践【SQL部分】2

public class FlinkCDCSQL {

	public static void main(String[] args) throws Exception {
		// 1.设置流的环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//		// 2.设置并行度
		env.setParallelism(1);
		StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

		//2.创建 Flink-MySQL-CDC 的 Source
		tableEnv.executeSql("CREATE TABLE student2 (" +
				" id STRING," +
				" name STRING," +
				" address STRING" +
				") WITH (" +
				" 'connector' = 'mysql-cdc'," +
				" 'scan.startup.mode' = 'latest-offset'," +
				" 'scan.incremental.snapshot.enabled' = 'false'," +
				" 'hostname' = 'localhost'," +
				" 'port' = '3306'," +
				" 'username' = 'root'," +
				" 'password' = 'root'," +
				" 'database-name' = 'test'," +
				" 'table-name' = 'student'" +
				")");
		tableEnv.executeSql("select * from student2").print();
		env.execute();
	}
}

Caused by: java.lang.NoSuchMethodError: org.apache.flink.table.factories.DynamicTableFactory$Context.getCatalogTable()Lorg/apache/flink/table/catalog/ResolvedCatalogTable;at com.ververica.cdc.connectors.mysql.table.MySqlTableSourceFactory.createDynamicTableSource(MySqlTableSourceFactory.java:80)
at org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:122)
... 19 more

Maven 将Flink版本换成
<flink-version>1.13.0</flink-version>

Caused by: org.apache.flink.table.api.ValidationException: The primary key is necessary when enable 'Key: 'scan.incremental.snapshot.enabled' , default: true (fallback keys: [])' to 'true'
	at com.ververica.cdc.connectors.mysql.table.MySqlTableSourceFactory.validatePrimaryKeyIfEnableParallel(MySqlTableSourceFactory.java:182)
	at com.ververica.cdc.connectors.mysql.table.MySqlTableSourceFactory.createDynamicTableSource(MySqlTableSourceFactory.java:85)
	at org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:128)
	... 19 more

再SQL语句中添加这一行，问题解决，1.13版本需要表有主键
" 'scan.incremental.snapshot.enabled' = 'false',"