[FLINK-34474][formats] Reset Avro decoder after a failed deserialization

[hulincup]

Problem fixed & how

AvroDeserializationSchema reuses a pooled MutableByteArrayInputStream and Decoder across messages. On each deserialize it swaps the input buffer via inputStream.setBuffer(message), then calls datumReader.read(null, decoder).

The JSON path reconfigures the JsonDecoder every call (((JsonDecoder) decoder).configure(inputStream)), but the binary path's BinaryDecoder is created once in checkAvroInitialized and never reconfigured afterwards — setBuffer only swaps the underlying stream, it does not touch the decoder's internal buffer.

When datumReader.read fails mid-record (e.g. a corrupt byte decodes to an out-of-range union tag → ArrayIndexOutOfBoundsException), the pooled BinaryDecoder keeps unconsumed bytes in its internal buffer. The next message swaps the input buffer, but the decoder's buffer is not cleared, so subsequent reads return corrupted data and every following message fails.

Reproducer (reported, binary encoding): valid → invalid → valid → valid becomes VALID → FAILED → FAILED → FAILED.

Behavior modified

• previous: after a failed deserialize (binary encoding), the pooled decoder stayed poisoned; every subsequent message failed regardless of content.
• now: on a failed read, the decoder is discarded and rebuilt bound to the current input stream, so the next message starts from a clean state. The original exception is rethrown unchanged.
• impact: failure path only; successful deserialization is unchanged.

Code refactored

deserialize wraps datumReader.read in a try-catch; on failure it calls a new resetDecoder() (rebuilds the JSON or binary decoder bound to the current input stream — binaryDecoder(inputStream, this.decoder) follows Avro's recommended reuse pattern) and then rethrows the original exception.

Features added / Functions optimized

N/A

Test plan

• Added testDeserializeRecoversFromCorruptMessage (parameterized over BINARY and JSON): a 2-branch union schema (string | int) where a multi-byte corrupt payload (leading byte 100 → zig-zag tag 50, out of range for 2 branches) throws mid-read and leaves trailing bytes in the decoder buffer; asserts valid → corrupt throws → valid still succeeds.
• The local environment runs Java 8 and cannot build Flink master (requires Java 11+), so verification relies on CI:

  mvn -pl flink-formats/flink-avro -am test -Dtest=AvroDeserializationSchemaTest
  

[flinkbot]

CI report:

• 412fa61 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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