DBZ-49 MySQL DDL parser should be more tolerant of REFERENCE clauses in CREATE TABLE statements

debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlDdlParser.java

@@ -578,7 +578,9 @@ protected void parseIndexOptions(Marker start) {
     protected void parseReferenceDefinition(Marker start) {
         tokens.consume("REFERENCES");
         parseSchemaQualifiedName(start); // table name
-        parseColumnNameList(start);
+        if (tokens.matches('(')) {
+            parseColumnNameList(start);
+        }
         if (tokens.canConsume("MATCH")) {
             tokens.consumeAnyOf("FULL", "PARTIAL", "SIMPLE");
             if (tokens.canConsume("ON")) {

debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/MySqlDdlParserTest.java

@@ -150,28 +150,44 @@ public void shouldParseGrantStatement() {
     public void shouldParseCreateStatements() {
         parser.parse(readFile("ddl/mysql-test-create.ddl"), tables);
         Testing.print(tables);
-        assertThat(tables.size()).isEqualTo(57); // no tables
+        assertThat(tables.size()).isEqualTo(57);
         assertThat(listener.total()).isEqualTo(144);
     }
 
     @Test
     public void shouldParseTestStatements() {
         parser.parse(readFile("ddl/mysql-test-statements.ddl"), tables);
         Testing.print(tables);
-        assertThat(tables.size()).isEqualTo(6); // no tables
+        assertThat(tables.size()).isEqualTo(6);
         assertThat(listener.total()).isEqualTo(49);
         //listener.forEach(this::printEvent);
     }
 
     @Test
     public void shouldParseSomeLinesFromCreateStatements() {
         parser.parse(readLines(189,"ddl/mysql-test-create.ddl"), tables);
-        assertThat(tables.size()).isEqualTo(39); // no tables
+        assertThat(tables.size()).isEqualTo(39);
         assertThat(listener.total()).isEqualTo(120);
     }
+
+    @Test
+    public void shouldParseMySql56InitializationStatements() {
+        parser.parse(readLines(1,"ddl/mysql-test-init-5.6.ddl"), tables);
+        assertThat(tables.size()).isEqualTo(85); // 1 table
+        assertThat(listener.total()).isEqualTo(112);
+        listener.forEach(this::printEvent);
+    }
+
+    @Test
+    public void shouldParseMySql57InitializationStatements() {
+        parser.parse(readLines(1,"ddl/mysql-test-init-5.7.ddl"), tables);
+        assertThat(tables.size()).isEqualTo(123);
+        assertThat(listener.total()).isEqualTo(125);
+        listener.forEach(this::printEvent);
+    }
 
     protected void printEvent( Event event ) {
-        System.out.println(event);
+        Testing.print(event);
     }
 
     protected String readFile( String classpathResource ) {

debezium-connector-mysql/src/test/java/io/debezium/connector/mysql/ReadBinLogIT.java

@@ -239,6 +239,29 @@ public void shouldCaptureMultipleWriteUpdateDeletesInSingleEvents() throws Excep
                                                  .removedRow("Jamie", 19, any(), any()));
     }
 
+    /**
+     * Test case that is normally commented out since it is only useful to print out the DDL statements recorded by
+     * the binlog during a MySQL server initialization and startup.
+     * 
+     * @throws Exception if there are problems
+     */
+    @Ignore
+    @Test
+    public void shouldCaptureQueryEventData() throws Exception {
+        // Testing.Print.enable();
+        startClient(client -> {
+            client.setBinlogFilename("mysql-bin.000001");
+            client.setBinlogPosition(4);
+        });
+        counters.consumeAll(5, TimeUnit.SECONDS);
+        List<QueryEventData> allQueryEvents = recordedEventData(QueryEventData.class, -1);
+        allQueryEvents.forEach(event -> {
+            String sql = event.getSql();
+            if (sql.equalsIgnoreCase("BEGIN") || sql.equalsIgnoreCase("COMMIT")) return;
+            System.out.println(event.getSql());
+        });
+    }
+
     @Test
     public void shouldQueryInformationSchema() throws Exception {
         // long tableId = writeRows.getTableId();