Unexpected exception while processing record 'ConsumerRecord(topic = qa-aurora-mysql-cluster.mysql.history, partition = 0, leaderEpoch = 2, offset = 2389, LogAppendTime = 1570538405771, serialized key size = -1, serialized value size = 674, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = {
  "source" : {
    "server" : "qa-aurora-mysql-cluster"
  },
  "position" : {
    "ts_sec" : 1570538405,
    "file" : "mysql-bin-changelog.002785",
    "pos" : 52763846,
    "gtids" : "c5bdbcc7-553c-3216-a74b-177e9033331f:1-79620029",
    "server_id" : 305502487
  },
  "databaseName" : "qa_proactive_messaging",
  "ddl" : "CREATE DEFINER=`qa_proact_rw`@`10.120.%` FUNCTION `uuidToBinary`(_uuid BINARY(36)) RETURNS binary(16)
    DETERMINISTIC
    SQL SECURITY INVOKER
RETURN
  UNHEX(CONCAT(
            SUBSTR(_uuid, 15, 4),
            SUBSTR(_uuid, 10, 4),
            SUBSTR(_uuid,  1, 8),
            SUBSTR(_uuid, 20, 4),
            SUBSTR(_uuid, 25) ))"
})'

io.debezium.text.ParsingException: no viable alternative at input 'CREATE DEFINER=`qa_proact_rw`@`10.120.%` FUNCTION `uuidToBinary`(_uuid BINARY(36)) RETURNS binary(16)\n    DETERMINISTIC\n    SQL SECURITY INVOKER\nRETURN'
	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:282)
	at org.antlr.v4.runtime.DefaultErrorStrategy.reportError(DefaultErrorStrategy.java:121)
	at io.debezium.ddl.parser.mysql.generated.MySqlParser.sqlStatements(MySqlParser.java:1078)
	at io.debezium.ddl.parser.mysql.generated.MySqlParser.root(MySqlParser.java:812)
	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.relational.history.AbstractDatabaseHistory.lambda$recover$1(AbstractDatabaseHistory.java:112)
	at io.debezium.relational.history.KafkaDatabaseHistory.recoverRecords(KafkaDatabaseHistory.java:248)
	at io.debezium.relational.history.AbstractDatabaseHistory.recover(AbstractDatabaseHistory.java:79)
	at io.debezium.connector.mysql.MySqlSchema.loadHistory(MySqlSchema.java:251)
	at io.debezium.connector.mysql.MySqlTaskContext.loadHistory(MySqlTaskContext.java:163)
	at io.debezium.connector.mysql.MySqlConnectorTask.start(MySqlConnectorTask.java:105)
	at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:49)
	at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:202)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	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: org.antlr.v4.runtime.NoViableAltException
	at org.antlr.v4.runtime.atn.ParserATNSimulator.noViableAlt(ParserATNSimulator.java:2023)
	at org.antlr.v4.runtime.atn.ParserATNSimulator.execATN(ParserATNSimulator.java:467)
	at org.antlr.v4.runtime.atn.ParserATNSimulator.adaptivePredict(ParserATNSimulator.java:393)
	at io.debezium.ddl.parser.mysql.generated.MySqlParser.sqlStatements(MySqlParser.java:886)
	... 19 more