FlinkCDC - mysql

1. 背景

FlinkSQL雙流關聯時，mysql流未能捕捉到Mysql表更新，所以嘗試使用FlinkCDC解決此問題。

2022-3-31 21:31:39 語法使用不當導致的未能捕捉，修改后可以。但是FlinkCDC研究下也沒壞處。

2. 代碼

2.1 使用

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DebeziumDeserializationSchema schema = new StringDebeziumDeserializationSchema();
        DebeziumSourceFunction function = MySQLSource.builder()
                .hostname("101.71.102.255")
                .port(8081)
                .databaseList("kafka")
                .username("kafka")
                .password("Bonc@123")
                .deserializer(schema)
                .build();

        env.addSource(function).print().setParallelism(1);

        env.execute();

3. 問題及解決辦法

3.1 無權限

Caused by: org.apache.kafka.connect.errors.ConnectException: Access denied; you need (at least one of) the RELOAD privilege(s) for this operation Error code: 1227; SQLSTATE: 42000.
	at io.debezium.connector.mysql.AbstractReader.wrap(AbstractReader.java:241)
	at io.debezium.connector.mysql.AbstractReader.failed(AbstractReader.java:218)
	at io.debezium.connector.mysql.SnapshotReader.execute(SnapshotReader.java:857)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.sql.SQLSyntaxErrorException: Access denied; you need (at least one of) the RELOAD privilege(s) for this operation
	at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:120)

mysql授權

grant reload on kafka.* to kafka@'%';
flush privilleges;