private JsonNode convert(ContainerNode<?> container, JsonNode reuse, TypeInformation<?> info, Object object) {
        if (info.equals(Types.VOID) || object == null) {
            return container.nullNode();
        } else if (info.equals(Types.BOOLEAN)) {
            return container.booleanNode((Boolean) object);
        } else if (info.equals(Types.STRING)) {
            return container.textNode((String) object);
        } else if (info.equals(Types.BIG_DEC)) {
            // convert decimal if necessary
            if (object instanceof BigDecimal) {
                return container.numberNode((BigDecimal) object);
            }
            return container.numberNode(BigDecimal.valueOf(((Number) object).doubleValue()));
        } else if (info.equals(Types.BIG_INT)) {
            // convert integer if necessary
            if (object instanceof BigInteger) {
                return container.numberNode((BigInteger) object);
            }
            return container.numberNode(BigInteger.valueOf(((Number) object).longValue()));
        } else if (info.equals(Types.SQL_DATE)) {
            return container.textNode(object.toString());
        } else if (info.equals(Types.SQL_TIME)) {
            final Time time = (Time) object;
            // strip milliseconds if possible
            if (time.getTime() % 1000 > 0) {
                return container.textNode(timeFormatWithMillis.format(time));
            }
            return container.textNode(timeFormat.format(time));
        } else if (info.equals(Types.SQL_TIMESTAMP)) {
            return container.textNode(timestampFormat.format((Timestamp) object));
        } else if (info instanceof RowTypeInfo) {
            if (reuse != null && reuse instanceof ObjectNode) {
                return convertRow((ObjectNode) reuse, (RowTypeInfo) info, (Row) object);
            } else {
                return convertRow(null, (RowTypeInfo) info, (Row) object);
            }
        } else if (info instanceof ObjectArrayTypeInfo) {
            if (reuse != null && reuse instanceof ArrayNode) {
                return convertObjectArray((ArrayNode) reuse, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object);
            } else {
                return convertObjectArray(null, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object);
            }
        } else if (info instanceof BasicArrayTypeInfo) {
            if (reuse != null && reuse instanceof ArrayNode) {
                return convertObjectArray((ArrayNode) reuse, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object);
            } else {
                return convertObjectArray(null, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object);
            }
        } else if (info instanceof PrimitiveArrayTypeInfo && ((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) {
            return container.binaryNode((byte[]) object);
        } else {
            // for types that were specified without JSON schema
            // e.g. POJOs
            try {
                return mapper.valueToTree(object);
            } catch (IllegalArgumentException e) {
                throw new IllegalStateException("Unsupported type information '" + info + "' for object: " + object, e);
            }
        }
    }
Creates a JSON serialization schema for the given type information.

@param typeInfo The field names of {@link Row} are used to map to JSON properties.
kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/JsonTupleSerializationSchema.java