DBZ-7718 Support table name escaping in BlockingSnapshotIT

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

@@ -137,6 +137,11 @@ protected String signalTableName() {
         return tableNameId("debezium_signal").toQuotedString('`');
     }
 
+    @Override
+    protected String escapedTableDataCollectionId() {
+        return String.format("\\\\\"%s\\\\\".\\\\\"%s\\\\\"", tableNameId().catalog(), tableNameId().table());
+    }
+
     @Override
     protected String signalTableNameSanitized() {
         return DATABASE.qualifiedTableName("debezium_signal");

debezium-connector-oracle/src/test/java/io/debezium/connector/oracle/BlockingSnapshotIT.java

@@ -110,6 +110,11 @@ protected String tableDataCollectionId() {
         return TestHelper.getDatabaseName() + ".DEBEZIUM.A";
     }
 
+    @Override
+    protected String escapedTableDataCollectionId() {
+        return "\\\"" + TestHelper.getDatabaseName() + "\\\".\\\"DEBEZIUM\\\".\\\"A\\\"";
+    }
+
     @Override
     protected List<String> tableDataCollectionIds() {
         return List.of(TestHelper.getDatabaseName() + ".DEBEZIUM.A", TestHelper.getDatabaseName() + ".DEBEZIUM.B");

debezium-connector-postgres/src/test/java/io/debezium/connector/postgresql/BlockingSnapshotIT.java

@@ -112,6 +112,11 @@ protected String signalTableName() {
         return "s1.debezium_signal";
     }
 
+    @Override
+    protected String escapedTableDataCollectionId() {
+        return "\\\"s1\\\".\\\"a\\\"";
+    }
+
     @Override
     protected String connector() {
         return "postgres";

debezium-connector-sqlserver/src/test/java/io/debezium/connector/sqlserver/BlockingSnapshotIT.java

@@ -88,6 +88,11 @@ protected String signalTableName() {
         return "dbo.debezium_signal";
     }
 
+    @Override
+    protected String escapedTableDataCollectionId() {
+        return "\\\"testDB1\\\".\\\"dbo\\\".\\\"a\\\"";
+    }
+
     @Override
     protected Configuration.Builder config() {
         return TestHelper.defaultConfig()

debezium-embedded/src/test/java/io/debezium/pipeline/AbstractBlockingSnapshotTest.java

@@ -57,6 +57,8 @@ public abstract class AbstractBlockingSnapshotTest extends AbstractSnapshotTest
     @Override
     protected abstract String tableName();
 
+    protected abstract String escapedTableDataCollectionId();
+
     @Override
     protected abstract String connector();
 
@@ -202,7 +204,7 @@ public void executeBlockingSnapshotWithEscapedCollectionName() throws Exception
         SourceRecords consumedRecordsByTopic = consumeRecordsByTopic(ROW_COUNT * 2, 10);
         assertRecordsFromSnapshotAndStreamingArePresent(ROW_COUNT * 2, consumedRecordsByTopic);
 
-        sendAdHocSnapshotSignalWithAdditionalConditionWithSurrogateKey("", "", BLOCKING, "\\\"s1\\\".\\\"a\\\"");
+        sendAdHocSnapshotSignalWithAdditionalConditionWithSurrogateKey("", "", BLOCKING, escapedTableDataCollectionId());
 
         waitForLogMessage("Snapshot completed", AbstractSnapshotChangeEventSource.class);