[FLINK-24544][formats] Fix Avro enum deserialization failure with Confluent Schema Registry

[nateab]

What is the purpose of the change

This pull request fixes a deserialization failure when using the Table API with Kafka + Avro + Confluent Schema Registry for records containing enum types. Deserialization fails with the error "Found MyEnumType, expecting union".

The root cause is that RegistryAvroFormatFactory derives a reader schema from the Table DDL via AvroSchemaConverter.convertToSchema(rowType). This conversion is lossy: Avro enums become Flink STRING, which converts back to Avro ["null", "string"] instead of ["null", {"type": "enum", ...}]. When Avro's GenericDatumReader performs schema resolution between the writer schema (from the registry, with enum) and the reader schema (from DDL, with string), it fails because union resolution cannot match an enum against a string.

The fix stops using the DDL-derived schema as the Avro reader schema when no explicit schema is provided via the avro-confluent.schema format option. Instead, the writer schema from the registry is used directly for deserialization. The AvroToRowDataConverters already handles enum-to-string conversion via .toString() at the Flink level, so Avro-level schema resolution is not needed for type coercion. When the user provides an explicit schema via avro-confluent.schema, it continues to be used as the reader schema.

Brief change log

• RegistryAvroFormatFactory: changed deserialization path to pass null instead of the DDL-derived schema when no avro-confluent.schema option is set (serialization path unchanged)
• AvroDeserializationSchema: checkAvroInitialized() now handles null schemaString gracefully; added Preconditions.checkNotNull guards in deserialize() and getProducedType() to fail fast with clear messages if null reader schema leaks into code paths that require it
• RegistryAvroDeserializationSchema: falls back to writer schema when reader schema is null via datumReader.setExpected(readerSchema != null ? readerSchema : writerSchema)
• Added test testRowDataReadWithEnumFieldAndNullReaderSchema in RegistryAvroRowDataSeDeSchemaTest
• Updated RegistryAvroFormatFactoryTest.testDeserializationSchema to match new null-schema behavior

Verifying this change

This change added tests and can be verified as follows:

• Added testRowDataReadWithEnumFieldAndNullReaderSchema that creates an Avro schema with a nullable enum field, serializes a GenericRecord using the Confluent wire format (magic byte + schema ID + Avro binary), then deserializes with a null reader schema and verifies the enum value is correctly read as a string
• Updated RegistryAvroFormatFactoryTest.testDeserializationSchema to expect null schema in the deserialization path
• Verified all existing tests pass: flink-avro (337 tests, 0 failures), flink-avro-confluent-registry (28 tests, 0 failures)

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes — the deserialization path now skips Avro schema resolution when no explicit reader schema is set, which is a slight performance improvement
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

[flinkbot]

CI report:

• ccdbdec Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[davidradl]

Would it be possible to put some of the tables mentioned in Jira as test cases to make sure they work and making it explicit under what circumstances this fix is required. For example does this effect sink cases as well as join cases?