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().equals(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 JSON schema.

@param jsonSchema JSON schema describing the result type

@see <a href="http://json-schema.org/">http://json-schema.org/</a>
kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/DTJsonRowSerializationSchema.java