private static SchemaAndValue parseAsExactDecimal(BigDecimal decimal) {
            BigDecimal ceil = decimal.setScale(0, RoundingMode.CEILING);
            BigDecimal floor = decimal.setScale(0, RoundingMode.FLOOR);
            if (ceil.equals(floor)) {
                BigInteger num = ceil.toBigIntegerExact();
                if (ceil.precision() >= 19 && (num.compareTo(LONG_MIN) < 0 || num.compareTo(LONG_MAX) > 0)) {
                    return null;
                }
                long integral = num.longValue();
                byte int8 = (byte) integral;
                short int16 = (short) integral;
                int int32 = (int) integral;
                if (int8 == integral) {
                    return new SchemaAndValue(Schema.INT8_SCHEMA, int8);
                } else if (int16 == integral) {
                    return new SchemaAndValue(Schema.INT16_SCHEMA, int16);
                } else if (int32 == integral) {
                    return new SchemaAndValue(Schema.INT32_SCHEMA, int32);
                } else {
                    return new SchemaAndValue(Schema.INT64_SCHEMA, integral);
                }
            }
            return null;
        }
Convert the specified value with the desired floating point type.

@param value  the value to be converted; may not be null
@param schema the schema for the current value type; may not be null
@param error  any previous error that should be included in an exception message; may be null
@return the double value after conversion; never null
@throws DataException if the value could not be converted to a double
connect/api/src/main/java/org/apache/kafka/connect/data/Values.java