This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 3160edda65d [FLINK-40263][format/avro] Validate enum symbols when
serializing CHAR/VARCHAR
3160edda65d is described below
commit 3160edda65d1b4b704314ac55421f0310b14ef67
Author: Pritam Kumar Mishra <[email protected]>
AuthorDate: Wed Jul 29 23:15:50 2026 +0530
[FLINK-40263][format/avro] Validate enum symbols when serializing
CHAR/VARCHAR
FLINK-39053 made RowDataToAvroConverters map a CHAR/VARCHAR column onto an
Avro
ENUM field, but it builds the EnumSymbol without checking that the string
is one
of the declared symbols. An unexpected value therefore survives the
conversion
and blows up further down inside GenericDatumWriter as 'Not an enum', which
names
neither the column nor the value nor the symbols that would have been
accepted.
Validate up front instead:
- a known symbol is converted as before;
- an unknown symbol falls back to the enum's declared default if it has
one,
which is what Avro's own schema resolution does when reading such a
value;
- otherwise the conversion fails with a message naming the value, the
enum and
the allowed symbols.
---
.../formats/avro/RowDataToAvroConverters.java | 24 +++++++++++-
.../AvroRowDataSchemaProvidedSerDeSchemaTest.java | 45 +++++++++++++++++++++-
2 files changed, 67 insertions(+), 2 deletions(-)
diff --git
a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RowDataToAvroConverters.java
b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RowDataToAvroConverters.java
index 202cd601bc0..63041161556 100644
---
a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RowDataToAvroConverters.java
+++
b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RowDataToAvroConverters.java
@@ -140,7 +140,9 @@ public class RowDataToAvroConverters {
@Override
public Object convert(Schema schema, Object
object) {
if (schema.getType() == Schema.Type.ENUM) {
- return new GenericData.EnumSymbol(schema,
object.toString());
+ // Returning a plain string here would
fail later on inside the
+ // datum writer, with no hint at the
offending field or value.
+ return convertToEnumSymbol(schema,
object.toString());
}
return new Utf8(object.toString());
}
@@ -299,6 +301,26 @@ public class RowDataToAvroConverters {
};
}
+ private static GenericData.EnumSymbol convertToEnumSymbol(Schema
enumSchema, String symbol) {
+ if (enumSchema.hasEnumSymbol(symbol)) {
+ return new GenericData.EnumSymbol(enumSchema, symbol);
+ }
+ // Mirror Avro's own schema resolution, which maps an unknown symbol
onto the declared
+ // default rather than failing.
+ final String fallback = enumSchema.getEnumDefault();
+ if (fallback != null) {
+ return new GenericData.EnumSymbol(enumSchema, fallback);
+ }
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot serialize '%s' as Avro enum '%s'. Allowed
symbols are [%s]. "
+ + "Declare a default symbol on the enum to map
unknown values onto "
+ + "it instead of failing.",
+ symbol,
+ enumSchema.getFullName(),
+ String.join(", ", enumSchema.getEnumSymbols())));
+ }
+
private static RowDataToAvroConverter createArrayConverter(
ArrayType arrayType, boolean legacyTimestampMapping) {
LogicalType elementType = arrayType.getElementType();
diff --git
a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDataSchemaProvidedSerDeSchemaTest.java
b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDataSchemaProvidedSerDeSchemaTest.java
index c07f153170a..7c6ae7d0ab1 100644
---
a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDataSchemaProvidedSerDeSchemaTest.java
+++
b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDataSchemaProvidedSerDeSchemaTest.java
@@ -172,7 +172,50 @@ class AvroRowDataSchemaProvidedSerDeSchemaTest {
assertThatThrownBy(() -> serializationSchema.serialize(rowData))
.isInstanceOf(RuntimeException.class)
- .hasMessageContaining("Failed to serialize row.");
+ .hasMessageContaining("Failed to serialize row.")
+ .hasStackTraceContaining("Fail to serialize at field:
type_enum")
+ .hasStackTraceContaining(
+ "Cannot serialize 'YELLOW' as Avro enum "
+ +
"'org.apache.flink.formats.avro.generated.Colors'")
+ .hasStackTraceContaining("Allowed symbols are [RED, GREEN,
BLUE]");
+ }
+
+ @ParameterizedTest
+ @EnumSource(AvroEncoding.class)
+ void testSerializeUnknownEnumSymbolFallsBackToEnumDefault(AvroEncoding
encoding)
+ throws Exception {
+ // An enum that declares a default is asking for unknown symbols to be
mapped onto it,
+ // which is what Avro's own schema resolution does when reading.
+ final Schema statusSchema =
+ new Schema.Parser()
+ .parse(
+ "{\"type\":\"enum\",\"name\":\"Status\","
+ +
"\"namespace\":\"org.apache.flink.formats.avro.generated\","
+ +
"\"symbols\":[\"ACTIVE\",\"INACTIVE\"],"
+ + "\"default\":\"INACTIVE\"}");
+ final Schema avroSchema =
+ SchemaBuilder.record("TestEnumDefault")
+ .namespace("org.apache.flink.formats.avro.generated")
+ .fields()
+ .name("status")
+ .type(statusSchema)
+ .noDefault()
+ .endRecord();
+
+ final RowType rowType = (RowType) ROW(FIELD("status",
STRING())).notNull().getLogicalType();
+
+ AvroRowDataSerializationSchema serializationSchema =
+ createSerializationSchema(rowType, avroSchema, encoding);
+ AvroRowDataDeserializationSchema deserializationSchema =
+ createDeserializationSchema(rowType, avroSchema, encoding);
+
+ GenericRowData rowData = new GenericRowData(1);
+ rowData.setField(0, StringData.fromString("ARCHIVED"));
+
+ RowData deserialized =
+
deserializationSchema.deserialize(serializationSchema.serialize(rowData));
+
+ assertThat(deserialized.getString(0).toString()).isEqualTo("INACTIVE");
}
@ParameterizedTest
