@Override
	public byte[] serialize(Tuple2<Boolean,Row> tuple2) {
		Row row = tuple2.f1;
		boolean change = tuple2.f0;
		if (root == null) {
			root = csvMapper.createObjectNode();
		}
		try {
			runtimeConverter.convert(csvMapper, root, row);
			if (StringUtils.equalsIgnoreCase(updateMode, EUpdateMode.UPSERT.name())) {
				fillRetractField(row, change);
			}

			return objectWriter.writeValueAsBytes(root);
		} catch (Throwable t) {
			throw new RuntimeException("Could not serialize row '" + row + "'.", t);
		}
	}
Creates a {@link CsvRowSerializationSchema} expecting the given {@link TypeInformation}.

@param typeInfo type information used to create schema.
kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/CsvTupleSerializationSchema.java