public static void fetchFieldNamesAndValuesFromIndexedRecord(IndexedRecord record, List<String> fieldNames,
      List<Object> fieldValues, Schema cachedSchema) {
    // Please note that record schema and cached schema could be different due to schema evolution.
    // Always represent record schema in the form of cached schema. This approach has the side-effect
    // of dropping the newly added fields in the scenarios where the record schema has newer version
    // than the cached schema. [TODO: SAMZA-1679]
    Schema recordSchema = record.getSchema();
    fieldNames.addAll(cachedSchema.getFields().stream().map(Schema.Field::name).collect(Collectors.toList()));
    fieldValues.addAll(fieldNames.stream()
        .map(f -> convertToJavaObject(
            recordSchema.getField(f) != null ? record.get(recordSchema.getField(f).pos()) : null,
            getNonNullUnionSchema(cachedSchema.getField(f).schema()))) // get schema from cachedSchema
        .collect(Collectors.toList()));
  }
Create a SamzaSqlRelMessage for the specified key and Avro record using the schema from the Avro record.
samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java