hulincup opened a new pull request, #28932:
URL: https://github.com/apache/flink/pull/28932
## 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
- [x] 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
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
