DBZ-7910 Fix column exclude filter type, add itest

src/main/java/io/debezium/connector/vitess/VitessConnectorConfig.java

@@ -489,7 +489,7 @@ public VitessConnectorConfig(Configuration config) {
                 config,
                 null, x -> x.schema() + "." + x.table(),
                 -1,
-                ColumnFilterMode.CATALOG,
+                ColumnFilterMode.SCHEMA,
                 true);
     }
 

src/test/java/io/debezium/connector/vitess/AbstractVitessConnectorTest.java

@@ -209,6 +209,22 @@ protected List<SchemaAndValueField> schemasAndValuesForStringTypesTruncated() {
         return fields;
     }
 
+    protected List<SchemaAndValueField> schemasAndValuesForStringTypesExcludedColumn() {
+        final List<SchemaAndValueField> fields = new ArrayList<>();
+        fields.addAll(
+                Arrays.asList(
+                        new SchemaAndValueField("char_col", SchemaBuilder.OPTIONAL_STRING_SCHEMA, "a"),
+                        new SchemaAndValueField("varchar_col", SchemaBuilder.OPTIONAL_STRING_SCHEMA, "bc"),
+                        new SchemaAndValueField("varchar_ko_col", SchemaBuilder.OPTIONAL_STRING_SCHEMA, "상품 명1"),
+                        new SchemaAndValueField("varchar_ja_col", SchemaBuilder.OPTIONAL_STRING_SCHEMA, "リンゴ"),
+                        new SchemaAndValueField("tinytext_col", SchemaBuilder.OPTIONAL_STRING_SCHEMA, "gh"),
+                        new SchemaAndValueField("text_col", SchemaBuilder.OPTIONAL_STRING_SCHEMA, "ij"),
+                        new SchemaAndValueField("longtext_col", SchemaBuilder.OPTIONAL_STRING_SCHEMA, "mn"),
+                        new SchemaAndValueField("json_col", Json.builder().optional().build(),
+                                "{\"key1\":\"value1\",\"key2\":{\"key21\":\"value21\",\"key22\":\"value22\"}}")));
+        return fields;
+    }
+
     protected List<SchemaAndValueField> schemasAndValuesForBytesTypesAsBytes() {
         final List<SchemaAndValueField> fields = new ArrayList<>();
         fields.addAll(

src/test/java/io/debezium/connector/vitess/VitessConnectorIT.java

@@ -249,6 +249,28 @@ public void shouldConsumeEventsWithTruncatedColumn() throws Exception {
         assertInsert(INSERT_STRING_TYPES_STMT, schemasAndValuesForStringTypesTruncated(), TestHelper.PK_FIELD);
     }
 
+    @Test
+    public void shouldConsumeEventsWithExcludedColumn() throws Exception {
+        String columnToExlude = "mediumtext_col";
+        String someColumnIncluded = "varchar_col";
+        TestHelper.executeDDL("vitess_create_tables.ddl");
+        startConnector(builder -> builder.with("column.exclude.list",
+                TEST_UNSHARDED_KEYSPACE + ".string_table." + columnToExlude), false,
+                false, 1, -1, -1, null,
+                VitessConnectorConfig.SnapshotMode.NEVER, "");
+        assertConnectorIsRunning();
+
+        int expectedRecordsCount = 1;
+        consumer = testConsumer(expectedRecordsCount);
+
+        consumer.expects(expectedRecordsCount);
+        SourceRecord record = assertInsert(INSERT_STRING_TYPES_STMT, schemasAndValuesForStringTypesExcludedColumn(), TestHelper.PK_FIELD);
+        Struct value = (Struct) record.value();
+        Struct after = (Struct) value.get("after");
+        assertThat(after.schema().field(columnToExlude)).isNull();
+        assertThat(after.schema().field(someColumnIncluded)).isNotNull();
+    }
+
     @Test
     public void shouldReceiveBytesAsBytes() throws Exception {
         TestHelper.executeDDL("vitess_create_tables.ddl");