DBZ-6185 Zerofill property failed for different int types

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

@@ -251,7 +251,7 @@ public void shouldProcessLargeColumn() {
     }
 
     @Test
-    @FixFor("DBZ-4497")
+    @FixFor({ "DBZ-4497", "DBZ-6185" })
     public void shouldProcessMultipleSignedUnsignedForTable() {
         String ddl = "create table if not exists tbl_signed_unsigned(\n"
                 + "`id` bigint(20) ZEROFILL signed UNSIGNED signed ZEROFILL unsigned ZEROFILL NOT NULL AUTO_INCREMENT COMMENT 'ID',\n"
@@ -259,6 +259,8 @@ public void shouldProcessMultipleSignedUnsignedForTable() {
                 + "c2 decimal(10, 2) SIGNED UNSIGNED ZEROFILL,\n"
                 + "c3 float SIGNED ZEROFILL,\n"
                 + "c4 double precision(18, 4) UNSIGNED SIGNED ZEROFILL,\n"
+                + "c5 smallint zerofill null,\n"
+                + "c6 tinyint unsigned zerofill null,\n"
                 + "PRIMARY KEY (`id`)\n"
                 + ")";
         parser.parse(ddl, tables);
@@ -275,12 +277,15 @@ public void shouldProcessMultipleSignedUnsignedForTable() {
         assertThat(c2.length()).isEqualTo(10);
         assertThat(c2.scale().get()).isEqualTo(2);
 
-        assertThat(table.columnWithName("c3").typeName()).isEqualTo("FLOAT SIGNED ZEROFILL");
+        assertThat(table.columnWithName("c3").typeName()).isEqualTo("FLOAT UNSIGNED ZEROFILL");
 
         Column c4 = table.columnWithName("c4");
         assertThat(c4.typeName()).isEqualTo("DOUBLE PRECISION UNSIGNED ZEROFILL");
         assertThat(c4.length()).isEqualTo(18);
         assertThat(c4.scale().get()).isEqualTo(4);
+
+        assertThat(table.columnWithName("c5").typeName()).isEqualTo("SMALLINT UNSIGNED ZEROFILL");
+        assertThat(table.columnWithName("c6").typeName()).isEqualTo("TINYINT UNSIGNED ZEROFILL");
     }
 
     @Test

debezium-core/src/main/java/io/debezium/relational/ddl/DataTypeBuilder.java

@@ -22,7 +22,8 @@ public class DataTypeBuilder {
     private int scale = -1;
     private int arrayDimsLength = 0;
     private final int[] arrayDims = new int[40];
-    private static final Pattern SIGNED_UNSIGNED_PATTERN = Pattern.compile("(.*)SIGNED UNSIGNED(.*)", Pattern.CASE_INSENSITIVE);
+    private static final Pattern SIGNED_UNSIGNED_ZEROFILL_PATTERN = Pattern
+            .compile("(.*)\\s+(SIGNED UNSIGNED ZEROFILL|SIGNED UNSIGNED|SIGNED ZEROFILL)", Pattern.CASE_INSENSITIVE);
 
     public void addToName(String str) {
         if (length == -1) {
@@ -105,19 +106,41 @@ else if (parameters != null) {
             name.append(' ');
             name.append(suffix);
         }
-        return new DataType(cleanSignedUnsigned(expression), cleanSignedUnsigned(name), jdbcType, length, scale, arrayDims, arrayDimsLength);
+        return new DataType(
+                adjustSignedUnsignedZerofill(expression),
+                adjustSignedUnsignedZerofill(name),
+                jdbcType,
+                length,
+                scale,
+                arrayDims,
+                arrayDimsLength);
     }
 
     /**
-     * Replace "signed unsigned" to "unsigned" when ddl statements contain multiple signed/unsigned keywords
+     * This method will adjust the suffix names of numeric data type.
+     * In connector streaming phase, the ddl parser maybe meet the invalid definition of suffix names with numeric data type,
+     * will adjust to appropriate values.
+     * e.g. replace "SIGNED UNSIGNED ZEROFILL" or "SIGNED ZEROFILL" to "UNSIGNED ZEROFILL", "SIGNED UNSIGNED" to "UNSIGNED"
+     * and adjust to "UNSIGNED ZEROFILL" if "zerofill" appears alone.
      */
-    private String cleanSignedUnsigned(StringBuilder origin) {
+    private String adjustSignedUnsignedZerofill(StringBuilder origin) {
-        Matcher matcher = SIGNED_UNSIGNED_PATTERN.matcher(origin.toString());
+        Matcher matcher = SIGNED_UNSIGNED_ZEROFILL_PATTERN.matcher(origin.toString());
         if (matcher.matches()) {
-            return matcher.replaceFirst("$1UNSIGNED$2");
+            String suffix = matcher.group(2).toUpperCase();
+            switch (suffix) {
+                case "SIGNED UNSIGNED ZEROFILL":
+                case "SIGNED ZEROFILL":
+                    return matcher.replaceFirst("$1 UNSIGNED ZEROFILL");
+                case "SIGNED UNSIGNED":
+                    return matcher.replaceFirst("$1 UNSIGNED");
+                default:
+                    return origin.toString();
+            }
         }
-        else {
+        if (origin.toString().toUpperCase().contains("ZEROFILL")
-            return origin.toString();
+                && !origin.toString().toUpperCase().contains("UNSIGNED")) {
+            return origin.toString().toUpperCase().replaceFirst("ZEROFILL", "UNSIGNED ZEROFILL");
         }
+        return origin.toString();
     }
 }

debezium-core/src/main/java/io/debezium/time/ZonedTimestamp.java

@@ -48,13 +48,13 @@ public class ZonedTimestamp {
     /**
      * Returns a {@link DateTimeFormatter} that ensures that exactly fractionalWidth number of digits are present
      * in the nanosecond part of the datetime. If fractionWidth is null, then
-     * {@link DateTimeFormatter.ISO_OFFSET_DATE_TIME} formatter is used, which can have anywhere from 0-9 digits in the
+     * {@link DateTimeFormatter#ISO_OFFSET_DATE_TIME} formatter is used, which can have anywhere from 0-9 digits in the
      * nanosecond part.
      *
      * @param fractionalWidth the optional component that specifies the exact number of digits to be present in a zoneddatetime
      *                        formatted string.
      * @return {@link DateTimeFormatter} containing exactly fractionalWidth number of digits in nanosecond part of the
-     * datetime. If null, {@link DateTimeFormatter.ISO_OFFSET_DATE_TIME} formatter is used, which can have anywhere
+     * datetime. If null, {@link DateTimeFormatter#ISO_OFFSET_DATE_TIME} formatter is used, which can have anywhere
      * from 0-9 digits in the nanosecond part.
      */
     private static DateTimeFormatter getDateTimeFormatter(Integer fractionalWidth) {