private ObjectNode convertRow(ObjectNode reuse, RowTypeInfo info, Row row) {
        if (reuse == null) {
            reuse = mapper.createObjectNode();
        }
        final String[] fieldNames = info.getFieldNames();

        final TypeInformation<?>[] fieldTypes = info.getFieldTypes();

        // validate the row
        if (row.getArity() != fieldNames.length) {
            throw new IllegalStateException(String.format(
                    "Number of elements in the row '%s' is different from number of field names: %d", row, fieldNames.length));
        }

        for (int i = 0; i < fieldNames.length; i++) {
            final String name = fieldNames[i];

            final JsonNode fieldConverted = convert(reuse, reuse.get(name), fieldTypes[i], row.getField(i));
            reuse.set(name, fieldConverted);
        }

        return reuse;
    }
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