NIFI-4844: Adjust BigDecimal scale to the target Avro schema

- Applied the same scale adjustment not only to BigDecimal inputs, but
also to Double values.

Signed-off-by: Matthew Burgess <mattyb149@apache.org>

This closes #2450

nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-avro-record-utils/src/main/java/org/apache/nifi/avro/AvroTypeUtil.java

@@ -19,7 +19,6 @@ package org.apache.nifi.avro;
 
 import java.io.IOException;
 import java.math.BigDecimal;
-import java.math.MathContext;
 import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
@@ -472,7 +471,7 @@ public class AvroTypeUtil {
 
     /**
      * Convert a raw value to an Avro object to serialize in Avro type system.
-     * The counter-part method which reads an Avro object back to a raw value is {@link #normalizeValue(Object, Schema)}.
+     * The counter-part method which reads an Avro object back to a raw value is {@link #normalizeValue(Object, Schema, String)}.
      */
     public static Object convertToAvroObject(final Object rawValue, final Schema fieldSchema) {
         return convertToAvroObject(rawValue, fieldSchema, fieldSchema.getName());
@@ -531,18 +530,18 @@ public class AvroTypeUtil {
                 final LogicalType logicalType = fieldSchema.getLogicalType();
                 if (logicalType != null && LOGICAL_TYPE_DECIMAL.equals(logicalType.getName())) {
                     final LogicalTypes.Decimal decimalType = (LogicalTypes.Decimal) logicalType;
-                    final BigDecimal decimal;
+                    final BigDecimal rawDecimal;
                     if (rawValue instanceof BigDecimal) {
-                        final BigDecimal rawDecimal = (BigDecimal) rawValue;
+                        rawDecimal = (BigDecimal) rawValue;
-                        final int desiredScale = decimalType.getScale();
-                        // If the desired scale is different than this value's coerce scale.
-                        decimal = rawDecimal.scale() == desiredScale ? rawDecimal : rawDecimal.setScale(desiredScale, BigDecimal.ROUND_HALF_UP);
                     } else if (rawValue instanceof Double) {
-                        // Scale is adjusted based on precision. If double was 123.456 and precision is 5, then decimal would be 123.46.
+                        rawDecimal = BigDecimal.valueOf((Double) rawValue);
-                        decimal = new BigDecimal((Double) rawValue, new MathContext(decimalType.getPrecision()));
                     } else {
                         throw new IllegalTypeConversionException("Cannot convert value " + rawValue + " of type " + rawValue.getClass() + " to a logical decimal");
                     }
+                    // If the desired scale is different than this value's coerce scale.
+                    final int desiredScale = decimalType.getScale();
+                    final BigDecimal decimal = rawDecimal.scale() == desiredScale
+                            ? rawDecimal : rawDecimal.setScale(desiredScale, BigDecimal.ROUND_HALF_UP);
                     return new Conversions.DecimalConversion().toBytes(decimal, fieldSchema, logicalType);
                 }
                 if (rawValue instanceof byte[]) {

nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-avro-record-utils/src/test/java/org/apache/nifi/avro/TestAvroTypeUtil.java

@@ -21,11 +21,17 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
+import org.apache.avro.Conversions;
+import org.apache.avro.LogicalTypes;
 import org.apache.avro.Schema;
 import org.apache.avro.Schema.Field;
 import org.apache.avro.Schema.Type;
@@ -259,4 +265,42 @@ public class TestAvroTypeUtil {
         }
     }
 
+    @Test
+    public void testToDecimalConversion() {
+        final LogicalTypes.Decimal decimalType = LogicalTypes.decimal(18, 8);
+        final Schema fieldSchema = Schema.create(Type.BYTES);
+        decimalType.addToSchema(fieldSchema);
+
+        final Map<Object, String> expects = new HashMap<>();
+
+        // Double to Decimal
+        expects.put(123d, "123.00000000");
+        // Double can not represent exact 1234567890.12345678, so use 1 less digit to test here.
+        expects.put(1234567890.12345678d, "1234567890.12345670");
+        expects.put(123456789.12345678d, "123456789.12345678");
+        expects.put(1234567890123456d, "1234567890123456.00000000");
+        // ROUND HALF UP.
+        expects.put(0.1234567890123456d, "0.12345679");
+
+
+        // BigDecimal to BigDecimal
+        expects.put(new BigDecimal("123"), "123.00000000");
+        expects.put(new BigDecimal("1234567890.12345678"), "1234567890.12345678");
+        expects.put(new BigDecimal("123456789012345678"), "123456789012345678.00000000");
+        // ROUND HALF UP.
+        expects.put(new BigDecimal("0.123456789012345678"), "0.12345679");
+
+
+        expects.forEach((rawValue, expect) -> {
+            final Object convertedValue = AvroTypeUtil.convertToAvroObject(rawValue, fieldSchema);
+
+            assertTrue(convertedValue instanceof ByteBuffer);
+            final ByteBuffer serializedBytes = (ByteBuffer) convertedValue;
+
+            final BigDecimal bigDecimal = new Conversions.DecimalConversion().fromBytes(serializedBytes, fieldSchema, decimalType);
+            assertEquals(String.format("%s should be converted to %s", rawValue, expect), expect, bigDecimal.toString());
+        });
+
+    }
+
 }