@Override
    public byte[] serialize(Tuple2<Boolean, Row> tuple2) {
        Row row = tuple2.f1;
        boolean change = tuple2.f0;
        if (node == null) {
            node = mapper.createObjectNode();
        }

        RowTypeInfo rowTypeInfo = (RowTypeInfo) ((TupleTypeInfo) typeInfo).getTypeAt(1);
        try {
            convertRow(node, rowTypeInfo, row);
            if (StringUtils.equalsIgnoreCase(updateMode, EUpdateMode.UPSERT.name())) {
                node.put(retractKey, change);
            }
            return mapper.writeValueAsBytes(node);
        } catch (Throwable t) {
            throw new RuntimeException("Could not serialize row '" + row + "'. " +
                    "Make sure that the schema matches the input.", t);
        }
    }
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