DBZ-3485 Emit quoted DROP TABLE

debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/antlr/listener/DropTableParserListener.java

@@ -33,7 +33,7 @@ public void enterDropTable(MySqlParser.DropTableContext ctx) {
         ctx.tables().tableName().forEach(tableNameContext -> {
             TableId tableId = parser.parseQualifiedTableId(tableNameContext.fullId());
             parser.databaseTables().removeTable(tableId);
-            parser.signalDropTable(tableId, prefix + tableId.table()
+            parser.signalDropTable(tableId, prefix + tableId.toQuotedString('`')
                     + (ctx.dropType != null ? " " + ctx.dropType.getText() : ""));
         });
         super.enterDropTable(ctx);

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

@@ -1079,7 +1079,7 @@ public void shouldParseMultipleStatements() {
         parser.parse(ddl, tables);
         assertThat(tables.size()).isEqualTo(0); // table created and dropped
         listener.assertNext().createTableNamed("foo").ddlStartsWith("CREATE TABLE foo (");
-        listener.assertNext().dropTableNamed("foo").ddlMatches("DROP TABLE foo");
+        listener.assertNext().dropTableNamed("foo").ddlMatches("DROP TABLE `foo`");
     }
 
     @Test

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

@@ -0,0 +1,78 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.connector.mysql;
+
+import java.nio.file.Path;
+import java.sql.SQLException;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import io.debezium.config.Configuration;
+import io.debezium.doc.FixFor;
+import io.debezium.embedded.AbstractConnectorTest;
+import io.debezium.util.Testing;
+
+/**
+*
+* The test to verify whether DDL is stored correctly in database history.
+*
+* @author Jiri Pechanec
+*/
+public class MySqlDatabaseHistoryIT extends AbstractConnectorTest {
+
+    private static final Path DB_HISTORY_PATH = Testing.Files.createTestingPath("file-db-history-json.txt")
+            .toAbsolutePath();
+
+    private static final int TABLE_COUNT = 1;
+
+    private UniqueDatabase DATABASE;
+
+    private Configuration config;
+
+    @Before
+    public void beforeEach() {
+        stopConnector();
+        DATABASE = new UniqueDatabase("history", "history-dbz")
+                .withDbHistoryPath(DB_HISTORY_PATH);
+        DATABASE.createAndInitialize();
+
+        initializeConnectorTestFramework();
+        Testing.Files.delete(DB_HISTORY_PATH);
+    }
+
+    @After
+    public void afterEach() {
+        try {
+            stopConnector();
+        }
+        finally {
+            Testing.Files.delete(DB_HISTORY_PATH);
+        }
+    }
+
+    @Test
+    @FixFor("DBZ-3485")
+    public void shouldConsumeAllEventsFromDatabaseUsingBinlogAndNoSnapshot() throws SQLException, InterruptedException {
+        config = DATABASE.defaultConfig()
+                .with(MySqlConnectorConfig.SNAPSHOT_MODE, MySqlConnectorConfig.SnapshotMode.SCHEMA_ONLY)
+                .build();
+
+        // Start the connector ...
+        start(MySqlConnector.class, config);
+
+        Testing.Print.enable();
+        // SET + USE + Drop DB + create DB + CREATE/DROP for each table
+        SourceRecords records = consumeRecordsByTopic(1 + 1 + 1 + 1 + TABLE_COUNT * 2);
+
+        stopConnector();
+
+        start(MySqlConnector.class, config);
+        assertConnectorIsRunning();
+        stopConnector();
+    }
+}

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

@@ -41,12 +41,12 @@ public class UniqueDatabase {
 
     private static final String DEFAULT_DATABASE = "mysql";
     private static final String[] CREATE_DATABASE_DDL = new String[]{
-            "CREATE DATABASE $DBNAME$;",
-            "USE $DBNAME$;"
+            "CREATE DATABASE `$DBNAME$`;",
+            "USE `$DBNAME$`;"
     };
     private static final String[] CREATE_DATABASE_WITH_CHARSET_DDL = new String[]{
-            "CREATE DATABASE $DBNAME$ CHARSET $CHARSET$;",
-            "USE $DBNAME$;"
+            "CREATE DATABASE `$DBNAME$` CHARSET $CHARSET$;",
+            "USE `$DBNAME$`;"
     };
     private static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)--.*$");
 

debezium-connector-mysql/src/test/resources/ddl/history-dbz.sql

@@ -0,0 +1,5 @@
+-- ----------------------------------------------------------------------------------------------------------------
+-- DATABASE:  history
+-- ----------------------------------------------------------------------------------------------------------------
+
+CREATE TABLE `t-1` (ID INT PRIMARY KEY);