chore: partial fix to stop ksql trimming trailing zeros from decimals

ksql-common/src/main/java/io/confluent/ksql/json/JsonMapper.java

@@ -18,6 +18,7 @@
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
 
 public enum JsonMapper {
@@ -32,5 +33,6 @@ public enum JsonMapper {
     mapper.registerModule(new KsqlTypesSerializationModule());
     mapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
     mapper.enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);
+    mapper.setNodeFactory(JsonNodeFactory.withExactBigDecimals(true));
   }
 }

ksql-common/src/test/java/io/confluent/ksql/json/JsonMapperTest.java

@@ -15,12 +15,16 @@
 
 package io.confluent.ksql.json;
 
-import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
 
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.DecimalNode;
+import java.math.BigDecimal;
 import org.junit.Test;
 
 public class JsonMapperTest {
@@ -34,6 +38,35 @@ public void shouldNotAutoCloseTarget() {
 
   @Test
   public void shouldIgnoreUnknownProperties() {
-    assertThat(OBJECT_MAPPER.isEnabled(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES), is(false));
+    assertThat(OBJECT_MAPPER.isEnabled(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES),
+        is(false));
+  }
+
+  @Test
+  public void shouldSerializeScientificDecimal() throws Exception {
+    // When:
+    final String json = OBJECT_MAPPER.writeValueAsString(new BigDecimal("1E+1"));
+
+    // Then:
+    assertThat(json, is("1E+1"));
+  }
+
+  @Test
+  public void shouldSerializeDecimalsWithoutLossOfTrailingZeros() throws Exception {
+    // When:
+    final String json = OBJECT_MAPPER.writeValueAsString(new BigDecimal("10.0"));
+
+    // Then:
+    assertThat(json, is("10.0"));
+  }
+
+  @Test
+  public void shouldDeserializeDecimalsWithoutLossOfTrailingZeros() throws Exception {
+    // When:
+    final JsonNode node = OBJECT_MAPPER.readTree("10.0");
+
+    // Then:
+    assertThat(node, is(instanceOf(DecimalNode.class)));
+    assertThat(node.decimalValue(), is(new BigDecimal("10.0")));
   }
 }

ksql-functional-tests/src/main/java/io/confluent/ksql/test/serde/json/ValueSpecJsonSerdeSupplier.java

@@ -39,7 +39,9 @@
 public class ValueSpecJsonSerdeSupplier implements SerdeSupplier<Object> {
 
   private static final ObjectMapper MAPPER = new ObjectMapper()
-      .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);
+      .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS)
+      .setNodeFactory(JsonNodeFactory.withExactBigDecimals(true));
+
   private final boolean useSchemas;
 
   public ValueSpecJsonSerdeSupplier(final boolean useSchemas) {
@@ -112,14 +114,17 @@ public Object deserialize(final String topicName, final byte[] data) {
 
   private static final class Converter<T> {
 
+    private static final JsonNodeFactory JSON_NODE_FACTORY = JsonNodeFactory
+        .withExactBigDecimals(true);
+
     private static final List<Converter<?>> CONVERTORS = ImmutableList.of(
-        Converter.converter(Boolean.class, JsonNodeFactory.instance::booleanNode),
-        Converter.converter(Integer.class, JsonNodeFactory.instance::numberNode),
-        Converter.converter(Long.class, JsonNodeFactory.instance::numberNode),
-        Converter.converter(Float.class, JsonNodeFactory.instance::numberNode),
-        Converter.converter(Double.class, JsonNodeFactory.instance::numberNode),
-        Converter.converter(BigDecimal.class, JsonNodeFactory.instance::numberNode),
-        Converter.converter(String.class, JsonNodeFactory.instance::textNode),
+        Converter.converter(Boolean.class, JSON_NODE_FACTORY::booleanNode),
+        Converter.converter(Integer.class, JSON_NODE_FACTORY::numberNode),
+        Converter.converter(Long.class, JSON_NODE_FACTORY::numberNode),
+        Converter.converter(Float.class, JSON_NODE_FACTORY::numberNode),
+        Converter.converter(Double.class, JSON_NODE_FACTORY::numberNode),
+        Converter.converter(BigDecimal.class, JSON_NODE_FACTORY::numberNode),
+        Converter.converter(String.class, JSON_NODE_FACTORY::textNode),
         Converter.converter(Collection.class, Converter::handleCollection),
         Converter.converter(Map.class, Converter::handleMap)
     );
@@ -129,7 +134,7 @@ private static final class Converter<T> {
 
     static JsonNode toJsonNode(final Object obj) {
       if (obj == null) {
-        return JsonNodeFactory.instance.nullNode();
+        return JSON_NODE_FACTORY.nullNode();
       }
 
       final List<Converter<?>> candidates = CONVERTORS.stream()
@@ -172,15 +177,15 @@ private JsonNode convert(final Object o) {
     }
 
     private static JsonNode handleCollection(final Collection<?> collection) {
-      final ArrayNode list = JsonNodeFactory.instance.arrayNode();
+      final ArrayNode list = JSON_NODE_FACTORY.arrayNode();
       for (final Object element : collection) {
         list.add(toJsonNode(element));
       }
       return list;
     }
 
     private static JsonNode handleMap(final Map<?, ?> map) {
-      final ObjectNode node = JsonNodeFactory.instance.objectNode();
+      final ObjectNode node = JSON_NODE_FACTORY.objectNode();
 
       if (map.isEmpty()) {
         return node;

ksql-functional-tests/src/main/java/io/confluent/ksql/test/tools/ExpectedRecordComparator.java

@@ -23,7 +23,6 @@
 import com.fasterxml.jackson.databind.node.TextNode;
 import com.google.common.collect.ImmutableMap;
 import java.math.BigDecimal;
-import java.math.RoundingMode;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -107,33 +106,47 @@ private static boolean compareArray(final Object actualValue, final JsonNode exp
 
   private static boolean compareNumber(final Object actualValue, final JsonNode expectedValue) {
     final NumericNode expected = (NumericNode) expectedValue;
+
     if (actualValue instanceof Integer) {
       return expected.intValue() == (Integer) actualValue;
     }
+
     if (actualValue instanceof Long) {
       return expected.longValue() == (Long) actualValue;
     }
+
     if (actualValue instanceof Double) {
       return expected.doubleValue() == (Double) actualValue;
     }
+
     if (actualValue instanceof BigDecimal) {
-      if (!expected.isBigDecimal()) {
-        // we don't want to risk comparing a BigDecimal with something of
-        // lower precision
-        return false;
-      }
+      return compareDecimal((BigDecimal) actualValue, expected);
+    }
+    return false;
+  }
 
-      try {
-        return expected.decimalValue()
-            .setScale(((BigDecimal) actualValue).scale(), RoundingMode.UNNECESSARY)
-            .equals(actualValue);
-      } catch (final ArithmeticException e) {
-        // the scale of the expected value cannot match the scale of the actual value
-        // without rounding
+  private static boolean compareDecimal(final BigDecimal actualValue, final NumericNode expected) {
+    if (expected.isInt() || expected.isLong()) {
+      // Only match if actual has no decimal places:
+      if (actualValue.scale() > 0) {
         return false;
       }
+
+      return expected.longValue() == actualValue.longValueExact();
+    }
+
+    if (!expected.isBigDecimal()) {
+      // we don't want to risk comparing a BigDecimal with something of lower precision
+      return false;
+    }
+
+    try {
+      return expected.decimalValue().equals(actualValue);
+    } catch (final ArithmeticException e) {
+      // the scale of the expected value cannot match the scale of the actual value
+      // without rounding
+      return false;
     }
-    return false;
   }
 
   private static boolean compareText(final Object actualValue, final JsonNode expectedValue) {

ksql-functional-tests/src/test/java/io/confluent/ksql/test/PlannedTestRewriterTest.java

@@ -20,6 +20,14 @@
 import org.junit.Test;
 
 public class PlannedTestRewriterTest {
+
+  /**
+   * Re-write ALL existing test plans.
+   *
+   * <p>You almost certainly do NOT want to do this as historical plans should be IMMUTABLE.
+   * The only time this is really valid is if you have fixed a bug in the testing framework
+   * and now need to correct bad historic test data.
+   */
   @Test
   @Ignore
   public void rewritePlans() {

ksql-functional-tests/src/test/java/io/confluent/ksql/test/serde/json/ValueSpecJsonSerdeSupplierTest.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2020 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.test.serde.json;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
+import io.confluent.ksql.json.JsonMapper;
+import io.confluent.ksql.serde.json.JsonSerdeUtils;
+import java.math.BigDecimal;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serializer;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ValueSpecJsonSerdeSupplierTest {
+
+  @Mock
+  private SchemaRegistryClient srClient;
+
+  private ValueSpecJsonSerdeSupplier plainSerde;
+  private ValueSpecJsonSerdeSupplier srSerde;
+
+  @Before
+  public void setUp() {
+    plainSerde = new ValueSpecJsonSerdeSupplier(false);
+    srSerde = new ValueSpecJsonSerdeSupplier(true);
+  }
+
+  @Test
+  public void shouldSerializeDecimalsWithOutStrippingTrailingZeros_Plain() {
+    // Given:
+    final Serializer<Object> serializer = plainSerde.getSerializer(srClient);
+
+    // When:
+    final byte[] bytes = serializer.serialize("t", new BigDecimal("10.0"));
+
+    // Then:
+    assertThat(new String(bytes, UTF_8), is("10.0"));
+  }
+
+  @Test
+  public void shouldDeserializeDecimalsWithoutStrippingTrailingZeros_Plain() {
+    // Given:
+    final Deserializer<Object> deserializer = plainSerde.getDeserializer(srClient);
+
+    final byte[] bytes = "10.0".getBytes(UTF_8);
+
+    // When:
+    final Object result = deserializer.deserialize("t", bytes);
+
+    // Then:
+    assertThat(result, is(new BigDecimal("10.0")));
+  }
+
+  @Test
+  public void shouldSerializeDecimalsWithOutStrippingTrailingZeros_Sr() throws Exception {
+    // Given:
+    final Serializer<Object> serializer = srSerde.getSerializer(srClient);
+
+    // When:
+    final byte[] bytes = serializer.serialize("t", new BigDecimal("10.0"));
+
+    // Then:
+    assertThat(JsonSerdeUtils.readJsonSR(bytes, JsonMapper.INSTANCE.mapper, String.class), is("10.0"));
+  }
+
+  @Test
+  public void shouldDeserializeDecimalsWithoutStrippingTrailingZeros_Sr() {
+    // Given:
+    final Deserializer<Object> deserializer = srSerde.getDeserializer(srClient);
+
+    final byte[] jsonBytes = "10.0".getBytes(UTF_8);
+    final byte[] bytes = new byte[jsonBytes.length + JsonSerdeUtils.SIZE_OF_SR_PREFIX];
+    System.arraycopy(jsonBytes, 0, bytes, JsonSerdeUtils.SIZE_OF_SR_PREFIX, jsonBytes.length);
+
+    // When:
+    final Object result = deserializer.deserialize("t", bytes);
+
+    // Then:
+    assertThat(result, is(new BigDecimal("10.0")));
+  }
+}