curl -H "Content-Type: application/json" \
 -X POST http://localhost:8083/connectors \
 -d '{
 "name": "mysql-connector",
 "config": {
 "connector.class": "io.debezium.connector.mysql.MySqlConnector",
 "database.serverTimezone": "Europe/Paris",
 "database.hostname": "localhost",
 "database.port": "3306",
 "database.user": "debezium",
 "database.password": "debezium",
 "database.server.id": "1",
 "database.server.name": "ZZZ_V3",
 "database.whitelist": "ZZZ_V3",
 "database.history.kafka.bootstrap.servers": "localhost:9092",
 "database.history.kafka.topic": "test"
 }
 }'

2020-07-08 16:57:10,243 INFO MySQL|ZZZ_V3|snapshot Step 7: rolling back transaction after abort [io.debezium.connector.mysql.SnapshotReader]
2020-07-08 16:57:10,244 ERROR MySQL|ZZZ_V3|snapshot Failed due to error: Aborting snapshot due to error when last running 'SHOW CREATE TABLE `mysql`.`global_priv`': no viable alternative at input 'CREATE TABLE `global_priv` (\n `Host` char(60) COLLATE utf8_bin NOT NULL DEFAULT '',\n `User` char(80) COLLATE utf8_bin NOT NULL DEFAULT '',\n `Priv` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL DEFAULT '{}' CHECK' [io.debezium.connector.mysql.SnapshotReader]
org.apache.kafka.connect.errors.ConnectException: no viable alternative at input 'CREATE TABLE `global_priv` (\n `Host` char(60) COLLATE utf8_bin NOT NULL DEFAULT '',\n `User` char(80) COLLATE utf8_bin NOT NULL DEFAULT '',\n `Priv` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL DEFAULT '{}' CHECK'
 at io.debezium.connector.mysql.AbstractReader.wrap(AbstractReader.java:230)
 at io.debezium.connector.mysql.AbstractReader.failed(AbstractReader.java:207)
 at io.debezium.connector.mysql.SnapshotReader.execute(SnapshotReader.java:831)
 at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
 at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
 at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: io.debezium.text.ParsingException: no viable alternative at input 'CREATE TABLE `global_priv` (\n `Host` char(60) COLLATE utf8_bin NOT NULL DEFAULT '',\n `User` char(80) COLLATE utf8_bin NOT NULL DEFAULT '',\n `Priv` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL DEFAULT '{}' CHECK'
 at io.debezium.antlr.ParsingErrorListener.syntaxError(ParsingErrorListener.java:40)
 at org.antlr.v4.runtime.ProxyErrorListener.syntaxError(ProxyErrorListener.java:41)
 at org.antlr.v4.runtime.Parser.notifyErrorListeners(Parser.java:544)
 at org.antlr.v4.runtime.DefaultErrorStrategy.reportNoViableAlternative(DefaultErrorStrategy.java:310)
 at org.antlr.v4.runtime.DefaultErrorStrategy.reportError(DefaultErrorStrategy.java:136)
 at io.debezium.ddl.parser.mysql.generated.MySqlParser.sqlStatements(MySqlParser.java:1121)
 at io.debezium.ddl.parser.mysql.generated.MySqlParser.root(MySqlParser.java:849)
 at io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser.parseTree(MySqlAntlrDdlParser.java:68)
 at io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser.parseTree(MySqlAntlrDdlParser.java:41)
 at io.debezium.antlr.AntlrDdlParser.parse(AntlrDdlParser.java:80)
 at io.debezium.connector.mysql.MySqlSchema.applyDdl(MySqlSchema.java:314)
 at io.debezium.connector.mysql.SnapshotReader.lambda$readTableSchema$15(SnapshotReader.java:864)
 at io.debezium.jdbc.JdbcConnection.query(JdbcConnection.java:473)
 at io.debezium.jdbc.JdbcConnection.query(JdbcConnection.java:414)
 at io.debezium.connector.mysql.SnapshotReader.readTableSchema(SnapshotReader.java:862)
 at io.debezium.connector.mysql.SnapshotReader.execute(SnapshotReader.java:524)

root@zzz02:/var/log/mysql# kafka-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic test | jq .

{
 "source": {
 "server": "ZZZ_V3"
 },
 "position": {
 "file": "mysql-bin.003104",
 "pos": 4403691,
 "snapshot": true
 },
 "databaseName": "mysql",
 "ddl": "CREATE TABLE `general_log` (\n `event_time` timestamp(6) NOT NULL DEFAULT current_timestamp(6) ON UPDATE current_timestamp(6),\n `user_host` mediumtext NOT NULL,\n `thread_id` bigint(21) unsigned NOT NULL,\n `server_id` int(10) unsigned NOT NULL,\n `command_type` varchar(64) NOT NULL,\n `argument` mediumtext NOT NULL\n) ENGINE=CSV DEFAULT CHARSET=utf8 COMMENT='General log'"
}