DBZ-3541 Round decimal default values of integer columns

debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlValueConverters.java

@@ -390,6 +390,16 @@ else if (data instanceof String) {
         });
     }
 
+    @Override
+    protected Object convertInteger(Column column, Field fieldDefn, Object data) {
+        // MySQL allows decimal default values for integer columns
+        if (data instanceof String) {
+            data = Math.round(Double.parseDouble((String) data));
+        }
+
+        return super.convertInteger(column, fieldDefn, data);
+    }
+
     /**
      * Convert the {@link String} or {@code byte[]} value to a string value used in a {@link SourceRecord}.
      *

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

@@ -529,6 +529,19 @@ public void shouldAcceptBitSetDefaultValue() {
         assertThat(Byte.toUnsignedInt((defVal[2]))).isEqualTo(0b11111111);
     }
 
+    @Test
+    @FixFor("DBZ-3541")
+    public void shouldRoundIntExpressedAsDecimal() {
+        String ddl = "CREATE TABLE int_as_decimal (col1 INT DEFAULT '0.0', col2 INT DEFAULT '1.5')";
+
+        parser.parse(ddl, tables);
+
+        Table table = tables.forTable(new TableId(null, null, "int_as_decimal"));
+
+        assertThat(getColumnSchema(table, "col1").defaultValue()).isEqualTo(0);
+        assertThat(getColumnSchema(table, "col2").defaultValue()).isEqualTo(2);
+    }
+
     @Test
     @FixFor("DBZ-3989")
     public void shouldTrimNumericalDefaultValueAndShouldNotTrimNonNumericalDefaultValue() {