Github user tragicjun commented on a diff in the pull request: https://github.com/apache/flink/pull/6082#discussion_r193168591 --- Diff: flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroRecordClassConverter.java --- @@ -73,9 +75,37 @@ private AvroRecordClassConverter() { final GenericTypeInfo<?> genericTypeInfo = (GenericTypeInfo<?>) extracted; if (genericTypeInfo.getTypeClass() == Utf8.class) { return BasicTypeInfo.STRING_TYPE_INFO; + } else if (genericTypeInfo.getTypeClass() == Map.class) { + // avro map key is always string + return Types.MAP(Types.STRING, + convertPrimitiveType(schema.getValueType().getType())); --- End diff -- I've implemented a reflection version, which now supports record type within map/array.
---