spratt opened a new pull request, #28715: URL: https://github.com/apache/flink/pull/28715
## What is the purpose of the change This PR addresses [FLINK-31951](https://issues.apache.org/jira/browse/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](https://github.com/apache/flink/pull/22507) (approved, closed inactive) and [#26397](https://github.com/apache/flink/pull/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](https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#7-testing). 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 -- 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]
