@Override
            public JsonNode toJson(final Schema schema, final Object value, final JsonConverterConfig config) {
                if (!(value instanceof BigDecimal))
                    throw new DataException("Invalid type for Decimal, expected BigDecimal but was " + value.getClass());

                final BigDecimal decimal = (BigDecimal) value;
                switch (config.decimalFormat()) {
                    case NUMERIC:
                        return JSON_NODE_FACTORY.numberNode(decimal);
                    case BASE64:
                        return JSON_NODE_FACTORY.binaryNode(Decimal.fromLogical(schema, decimal));
                    default:
                        throw new DataException("Unexpected " + JsonConverterConfig.DECIMAL_FORMAT_CONFIG + ": " + config.decimalFormat());
                }
            }
Implementation of {@link Converter} and {@link HeaderConverter} that uses JSON to store schemas and objects. By
default this converter will serialize Connect keys, values, and headers with schemas, although this can be disabled with
the {@link JsonConverterConfig#SCHEMAS_ENABLE_CONFIG schemas.enable} configuration option.
<p>
This implementation currently does nothing with the topic names or header keys.
connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java