[Decimal] Fix Encoding Problem (#734)

lib/debezium/decimal.go

@@ -9,15 +9,22 @@ import (
 
 // EncodeDecimal is used to encode a string representation of a number to `org.apache.kafka.connect.data.Decimal`.
 func EncodeDecimal(value string, scale int) ([]byte, error) {
-	scaledValue := new(big.Int).Exp(big.NewInt(10), big.NewInt(int64(scale)), nil)
+	// TODO: Refactor scale to be uint16
+
 	bigFloatValue := new(big.Float)
 	if _, success := bigFloatValue.SetString(value); !success {
-		return nil, fmt.Errorf("unable to use '%s' as a floating-point number", value)
+		return nil, fmt.Errorf("unable to use %q as a floating-point number", value)
 	}
+
+	scaledValue := new(big.Int).Exp(big.NewInt(10), big.NewInt(int64(scale)), nil)
 	bigFloatValue.Mul(bigFloatValue, new(big.Float).SetInt(scaledValue))
 
 	// Extract the scaled integer value.
-	bigIntValue, _ := bigFloatValue.Int(nil)
+	bigIntValue := new(big.Int)
+	if _, success := bigIntValue.SetString(bigFloatValue.String(), 10); !success {
+		return nil, fmt.Errorf("unable to use %q as a floating-point number", value)
+	}
+
 	data := bigIntValue.Bytes()
 	if bigIntValue.Sign() < 0 {
 		// Convert to two's complement if the number is negative

lib/debezium/decimal_test.go

@@ -66,15 +66,20 @@ func TestEncodeDecimal(t *testing.T) {
 			value: "6408.355",
 			scale: 3,
 		},
+		{
+			name:  "total",
+			value: "1.05",
+			scale: 2,
+		},
 		{
 			name:        "malformed - empty string",
 			value:       "",
-			expectedErr: "unable to use '' as a floating-point number",
+			expectedErr: `unable to use "" as a floating-point number`,
 		},
 		{
 			name:        "malformed - not a floating-point",
 			value:       "abcdefg",
-			expectedErr: "unable to use 'abcdefg' as a floating-point number",
+			expectedErr: `unable to use "abcdefg" as a floating-point number`,
 		},
 	}