[FLINK-31951][format] Reset decoder on Avro deserialization failure

[spratt]

What is the purpose of the change

This PR addresses FLINK-31951. We are encountering this bug in production and opened this PR after finding the two prior fix attempts closed due to inactivity: #22507 (approved, closed inactive) and #26397 (closed inactive).

This fix was developed independently. The prior PRs applied the reset in AvroDeserializationSchema.deserialize(), but RegistryAvroDeserializationSchema overrides that method entirely, so the base-class fix is never reached when using the registry schema. This PR targets RegistryAvroDeserializationSchema.deserialize() directly and also catches RuntimeException (covering AvroRuntimeException, the exception type thrown when the decoder reads malformed data). We are happy to defer if the maintainers prefer to revive one of the earlier approaches instead.

RegistryAvroDeserializationSchema reuses a single BinaryDecoder instance across all messages on a Kafka partition. When datumReader.read() fails mid-message (e.g. a malformed payload, a schema mismatch, or an AvroRuntimeException), stale bytes remain in the decoder's internal read-ahead buffer. The next call to datumReader.read() then starts from those leftover bytes rather than the beginning of the next message, producing corrupt output or cascading failures.

This fix wraps datumReader.read() in a try-catch and calls resetDecoder() before re-throwing, discarding any pre-fetched bytes and leaving the decoder ready for the next message.

Brief change log

• AvroDeserializationSchema: add package-private resetDecoder() method that reinitialises the BinaryDecoder against the existing inputStream via DecoderFactory.get().binaryDecoder(inputStream, decoder); no-op for JSON encoding
• RegistryAvroDeserializationSchema: wrap datumReader.read() in a try-catch on IOException | RuntimeException; call resetDecoder() before re-throwing so that a failed decode does not corrupt the bytes available to the next message
• Add RegistryAvroDeserializationSchemaDecoderResetTest: serialises a malformed message followed by a valid one and asserts the valid message deserialises correctly after the failed decode

Verifying this change

Please make sure both new and modified tests in this PR follow the conventions for tests defined in our code quality guide.

This change added tests and can be verified as follows:

• Added RegistryAvroDeserializationSchemaDecoderResetTest which writes a Confluent Schema Registry framed message with a garbage Avro payload (triggering an AvroRuntimeException mid-decode), followed by a correctly encoded message, then deserialises both in sequence and asserts that the second message produces the expected field values (id=42, name="hello"). Without the fix, the second decode reads from the stale bytes left by the first failure and also throws.

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 (resetDecoder() is package-private; no public method signatures are changed)
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes; datumReader.read() is now wrapped in a try-catch on every deserialization call. In the happy path (no exception thrown) this has negligible overhead in modern JVMs
• Anything that affects deployment or recovery: 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:

• 07262e0 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[spratt]

@flinkbot run azure