[hotfix][cdc-connector][mysql] Fix the default length of BINARY data type of Add column DDL

Author: zhongqs

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/utils/MySqlTypeUtils.java

@@ -218,7 +218,7 @@ private static DataType convertFromColumn(Column column) {
             case MULTILINESTRING:
                 return DataTypes.STRING();
             case BINARY:
-                return DataTypes.BINARY(column.length());
+                return DataTypes.BINARY(column.length() > 0 ? column.length() : 1);
             case VARBINARY:
                 return DataTypes.VARBINARY(column.length());
             case TINYBLOB:

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java

@@ -301,6 +301,16 @@ public void testParseAlterStatement() throws Exception {
                             Collections.singletonList(
                                     new AddColumnEvent.ColumnWithPosition(
                                             Column.physicalColumn("cols5", DataTypes.BOOLEAN())))));
+            statement.execute(
+                    String.format(
+                            "ALTER TABLE `%s`.`products` ADD COLUMN `cols6` BINARY NULL;",
+                            inventoryDatabase.getDatabaseName()));
+            expected.add(
+                    new AddColumnEvent(
+                            tableId,
+                            Collections.singletonList(
+                                    new AddColumnEvent.ColumnWithPosition(
+                                            Column.physicalColumn("cols6", DataTypes.BINARY(1))))));
         }
         List<Event> actual = fetchResults(events, expected.size());
         assertThat(actual).isEqualTo(expected);