From bd4f3d60faad3af245d7ce307fd581611093a4e3 Mon Sep 17 00:00:00 2001
From: Gunnar Morling <gunnar.morling@googlemail.com>
Date: Tue, 24 Jul 2018 11:24:44 +0200
Subject: [PATCH] DBZ-804 Simplifying construction of VariableScaleDecimal

---
 .../debezium/data/VariableScaleDecimal.java   | 25 +++++++++++++++----
 1 file changed, 20 insertions(+), 5 deletions(-)

diff --git a/debezium-core/src/main/java/io/debezium/data/VariableScaleDecimal.java b/debezium-core/src/main/java/io/debezium/data/VariableScaleDecimal.java
index 27a659eff..c804a77d5 100644
--- a/debezium-core/src/main/java/io/debezium/data/VariableScaleDecimal.java
+++ b/debezium-core/src/main/java/io/debezium/data/VariableScaleDecimal.java
@@ -8,6 +8,7 @@
 
 import java.math.BigDecimal;
 import java.math.BigInteger;
+import java.util.Objects;
 
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.SchemaBuilder;
@@ -15,7 +16,7 @@
 
 /**
  * An arbitrary precision decimal value with variable scale.
- * 
+ *
  * @author Jiri Pechanec
  *
  */
@@ -61,7 +62,7 @@ public static Schema optionalSchema() {
     }
 
     /**
-     * Converts a value from its logical format (BigDecimal) to its encoded format - a struct containing
+     * Converts a value from its logical format to its encoded format - a struct containing
      * the scale of the number and a binary representation of the number.
      *
      * @param schema of the encoded value
@@ -70,11 +71,25 @@ public static Schema optionalSchema() {
      * @return the encoded value
      */
     public static Struct fromLogical(Schema schema, SpecialValueDecimal value) {
+        return fromLogical(schema, value.getDecimalValue().orElse(null));
+    }
+
+    /**
+     * Converts a value from its logical format to its encoded format - a struct containing
+     * the scale of the number and a binary representation of the number.
+     *
+     * @param schema of the encoded value
+     * @param value the value or the decimal
+     *
+     * @return the encoded value
+     */
+    public static Struct fromLogical(Schema schema, BigDecimal decimalValue) {
+        Objects.requireNonNull(decimalValue, "decimalValue may not be null");
+
         Struct result = new Struct(schema);
-        final BigDecimal decimalValue = value.getDecimalValue().orElse(null);
-        assert decimalValue != null : "Unable to encode special value";
         result.put(VALUE_FIELD, decimalValue.unscaledValue().toByteArray());
         result.put(SCALE_FIELD, decimalValue.scale());
+
         return result;
     }
 
@@ -85,6 +100,6 @@ public static Struct fromLogical(Schema schema, SpecialValueDecimal value) {
      * @return the decoded value
      */
     public static SpecialValueDecimal toLogical(final Struct value) {
-        return new SpecialValueDecimal(new BigDecimal(new BigInteger((byte[])value.getBytes(VALUE_FIELD)), value.getInt32(SCALE_FIELD)));
+        return new SpecialValueDecimal(new BigDecimal(new BigInteger(value.getBytes(VALUE_FIELD)), value.getInt32(SCALE_FIELD)));
     }
 }