private GenericRecord convertRowToAvroRecord(Schema schema, Row row) {

		final List<Schema.Field> fields = schema.getFields()
				.stream()
				.filter(field -> !StringUtils.equalsIgnoreCase(field.name(), retractKey))
				.collect(Collectors.toList());

		final int length = fields.size();
		final GenericRecord record = new GenericData.Record(schema);
		for (int i = 0; i < length; i++) {
			final Schema.Field field = fields.get(i);
			record.put(i, convertFlinkType(field.schema(), row.getField(i)));
		}
		return record;
	}
Creates an Avro serialization schema for the given Avro schema string.

@param avroSchemaString Avro schema string used to serialize Flink's row to Avro's record
kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/AvroTuple2SerializationSchema.java