private Schema computePayloadSchema(String streamName, SamzaSqlRelMessage relMessage) {
    SamzaSqlRelRecord relRecord = relMessage.getSamzaSqlRelRecord();
    List<Schema.Field> keyFields = new ArrayList<>();
    List<String> fieldNames = relRecord.getFieldNames();
    List<Object> values = relRecord.getFieldValues();

    for (int index = 0; index < fieldNames.size(); index++) {
      if (fieldNames.get(index).equals(SamzaSqlRelMessage.KEY_NAME) || values.get(index) == null) {
        continue;
      }

      Object value = values.get(index);
      Schema avroType;
      if (value instanceof GenericData.Record) {
        avroType = ((GenericData.Record) value).getSchema();
      } else {
        avroType = ReflectData.get().getSchema(value.getClass());
      }
      keyFields.add(new Schema.Field(fieldNames.get(index), avroType, "", null));
    }

    Schema ks = Schema.createRecord(streamName, "", streamName + "_namespace", false);
    ks.setFields(keyFields);
    String schemaStr = ks.toString();
    Schema schema;
    // See whether we have a schema object corresponding to the schemaValue and reuse it.
    // CachedSchemaRegistryClient doesn't like if we recreate schema objects.
    if (schemas.containsKey(schemaStr)) {
      schema = schemas.get(schemaStr);
    } else {
      schema = Schema.parse(schemaStr);
      schemas.put(schemaStr, schema);
    }

    return schema;
  }
Special form for AvroRelConverter that generates the avro schema on the output based on the
fields in {@link SamzaSqlRelMessage} and uses the schema to serialize the output.
This is useful to test out the SQL quickly when the destination system supports Avro serialized data,
without having to manually author the avro schemas for various SQL queries.
samza-tools/src/main/java/org/apache/samza/tools/avro/AvroSchemaGenRelConverter.java