[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 extends the fix to cover both: it adds the same try-catch to AvroDeserializationSchema.deserialize() (addressing the base-class bug) and separately to RegistryAvroDeserializationSchema.deserialize() (which cannot rely on the base-class fix since it overrides the method). Both catch RuntimeException in addition to IOException, 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.

AvroDeserializationSchema and its subclass RegistryAvroDeserializationSchema both reuse 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 in both classes 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; wrap datumReader.read() in a try-catch on IOException | RuntimeException that calls resetDecoder() before re-throwing
• RegistryAvroDeserializationSchema: wrap datumReader.read() in the same try-catch; this is necessary independently of the base-class fix because the registry class overrides deserialize() entirely
• Add AvroDeserializationSchemaDecoderResetTest: same pattern with raw Avro binary (no registry header), verifying the base-class fix
• Add RegistryAvroDeserializationSchemaDecoderResetTest: serialises a Confluent-framed 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 AvroDeserializationSchemaDecoderResetTest which feeds a raw Avro binary payload with a valid id field but an invalid string length for name (triggering an AvroRuntimeException mid-decode), followed by a correctly encoded message, and asserts the second message produces id=42, name="hello". Without the base-class fix, the second decode reads the stale zero bytes left by the first failure and returns id=0, name="".
• Added RegistryAvroDeserializationSchemaDecoderResetTest which does the same with Confluent Schema Registry framed messages (magic byte + schema ID prefix). Without the registry fix, the second decode similarly returns wrong values from the stale buffer, demonstrating that the registry class must be fixed independently.

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:

• c1dfa93 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[spratt]

@flinkbot run azure

[davidradl]

this looks like a great catch and one to get fixed. I am curious whether this scenario can be hit in the base avro format itself - if so could we fix it there?

[spratt]

Great question. I'll investigate and get back to you.

[spratt]

Well caught! My investigation showed that the scenario could be hit independently in the base class and in the registry-specific case, so I've included tests and fixes for both.