fix: zero decimal bug

ksqldb-common/src/main/java/io/confluent/ksql/util/DecimalUtil.java

@@ -234,6 +234,14 @@ private static void ensureMax(final BigDecimal value, final int precision, final
   }
 
   public static SqlType fromValue(final BigDecimal value) {
+    final BigDecimal bigDecimalZero = BigDecimal.ZERO;
+
+    /* When a BigDecimal has value 0, the built-in method precision() always returns 1. To account
+    for this edge case, we just take the scale and add one and use that for the precision instead.
+    */
+    if (value.compareTo(bigDecimalZero) == 0) {
+      return SqlTypes.decimal(value.scale() + 1, value.scale());
+    }
     return SqlTypes.decimal(value.precision(), value.scale());
   }
 }

ksqldb-common/src/test/java/io/confluent/ksql/util/DecimalUtilTest.java

@@ -85,6 +85,14 @@ public void shouldExtractPrecisionFromDecimalSchema() {
     assertThat(scale, is(2));
   }
 
+  @Test
+  public void shouldExtractPrecisionFromZeroValue() {
+    // When:
+    final SqlType zeroDecimal = DecimalUtil.fromValue(BigDecimal.ZERO.setScale(2));
+
+    // Then:
+    assertThat(zeroDecimal, is(SqlTypes.decimal(3,2)));
+  }
   @Test
   public void shouldCastDecimal() {
     // When:

ksqldb-functional-tests/src/test/resources/rest-query-validation-tests/insert-values.json

@@ -344,6 +344,18 @@
       "outputs": [
         {"topic": "test_topic", "key": null, "value": {"ID": 1}}
       ]
+    },
+    {
+      "name": "should insert decimals that equal zero",
+      "statements": [
+        "CREATE STREAM S (ID INT KEY, V0 DECIMAL(12, 2))  WITH (kafka_topic='test_topic',partitions = 1, value_format='JSON');",
+        "INSERT INTO S (ID,V0) VALUES (1,0.00);"
+      ],
+      "inputs": [
+      ],
+      "outputs": [
+        {"topic": "test_topic", "key": 1, "value": {"V0": 0.00}}
+      ]
     }
   ]
 }