From 58fed0058194ce4c9174a19927abf621bc912a21 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=83=A1=E7=90=B3=28HuLin=29?= Date: Thu, 6 Aug 2026 12:21:58 +0800 Subject: [PATCH 1/3] [FLINK-34474][formats] Reset Avro decoder after a failed deserialization AvroDeserializationSchema reuses a pooled MutableByteArrayInputStream and Decoder across messages. On each deserialize it swaps the input buffer via setBuffer, then calls datumReader.read(null, decoder). The JSON path reconfigures the JsonDecoder every call, but the binary path's BinaryDecoder is created once and never reconfigured. When datumReader.read fails mid-record (e.g. a corrupt byte decoding to an out-of-range union tag throws 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. Reproduce (reported, binary encoding): valid -> invalid -> valid -> valid becomes VALID -> FAILED -> FAILED -> FAILED. On a failed read, discard the poisoned decoder and rebuild it bound to the current input stream (binaryDecoder reuses the existing decoder per Avro's recommended reuse pattern), then rethrow the original exception. Adds testDeserializeRecoversFromCorruptMessage: a 2-branch union schema where a multi-byte corrupt payload (leading byte 100 -> zig-zag tag 50, out of range) throws mid-read and leaves trailing bytes in the decoder buffer; asserts a valid message still deserializes afterwards. --- .../avro/AvroDeserializationSchema.java | 25 +++++++++++- .../avro/AvroDeserializationSchemaTest.java | 39 +++++++++++++++++++ 2 files changed, 63 insertions(+), 1 deletion(-) diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroDeserializationSchema.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroDeserializationSchema.java index 226f47a339b213..e07a0f3cbe9b81 100644 --- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroDeserializationSchema.java +++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroDeserializationSchema.java @@ -181,7 +181,30 @@ public T deserialize(@Nullable byte[] message) throws IOException { ((JsonDecoder) this.decoder).configure(inputStream); } - return datumReader.read(null, decoder); + try { + return datumReader.read(null, decoder); + } catch (IOException | RuntimeException e) { + // FLINK-34474: a failed read can leave the pooled decoder in an + // inconsistent internal state, so that subsequent reads return + // corrupted data even after the input buffer is reset. Discard + // the poisoned decoder so the next message starts from a clean state. + resetDecoder(); + throw e; + } + } + + private void resetDecoder() { + try { + if (encoding == AvroEncoding.JSON) { + this.decoder = DecoderFactory.get().jsonDecoder(getReaderSchema(), inputStream); + } else { + this.decoder = DecoderFactory.get().binaryDecoder(inputStream, this.decoder); + } + } catch (IOException e) { + // jsonDecoder only throws on schema/input issues that cannot occur here + // (the schema is cached and the input is an in-memory stream). + throw new RuntimeException(e); + } } void checkAvroInitialized() throws IOException { diff --git a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroDeserializationSchemaTest.java b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroDeserializationSchemaTest.java index fd0d05ffa4bc97..456c4a5594f675 100644 --- a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroDeserializationSchemaTest.java +++ b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroDeserializationSchemaTest.java @@ -24,15 +24,19 @@ import org.apache.flink.formats.avro.generated.UnionLogicalType; import org.apache.flink.formats.avro.utils.TestDataGenerator; +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; import org.apache.avro.generic.GenericRecord; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; import java.time.Instant; +import java.util.Collections; import java.util.Random; import static org.apache.flink.formats.avro.utils.AvroTestUtils.writeRecord; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link AvroDeserializationSchema}. */ class AvroDeserializationSchemaTest { @@ -85,4 +89,39 @@ void testSpecificRecordWithUnionLogicalType(AvroEncoding encoding) throws Except UnionLogicalType deserializedData = deserializer.deserialize(encodedData); assertThat(deserializedData).isEqualTo(data); } + + @ParameterizedTest + @EnumSource(AvroEncoding.class) + void testDeserializeRecoversFromCorruptMessage(AvroEncoding encoding) throws Exception { + // Schema with a 2-branch union so a corrupt tag triggers an + // ArrayIndexOutOfBoundsException mid-read (FLINK-34474 reproducer). + Schema schema = Schema.createRecord("corruptTest", null, null, false); + schema.setFields( + Collections.singletonList( + new Schema.Field( + "f", + Schema.createUnion( + Schema.create(Schema.Type.STRING), + Schema.create(Schema.Type.INT))))); + + DeserializationSchema deserializer = + AvroDeserializationSchema.forGeneric(schema, encoding); + + GenericRecord valid = new GenericData.Record(schema); + valid.put("f", "hello"); + byte[] validBytes = writeRecord(valid, schema, encoding); + + // The leading byte (100) decodes to an out-of-range union tag (zig-zag 50) + // and throws mid-read; the trailing bytes are left unconsumed in the + // pooled BinaryDecoder's internal buffer, poisoning subsequent reads + // unless the decoder is reset (FLINK-34474). + byte[] corrupt = new byte[] {100, 0, 0, 0, 0, 0, 0, 0}; + + assertThat(deserializer.deserialize(validBytes)).isEqualTo(valid); + assertThatThrownBy(() -> deserializer.deserialize(corrupt)) + .isInstanceOf(Exception.class); + // FLINK-34474: a subsequent valid message must still deserialize, + // instead of being poisoned by the prior failed read. + assertThat(deserializer.deserialize(validBytes)).isEqualTo(valid); + } } From 5617104cd252c09a990b2fb52ea1c1e52a5535d7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=83=A1=E7=90=B3=28HuLin=29?= Date: Thu, 6 Aug 2026 13:56:45 +0800 Subject: [PATCH 2/3] [FLINK-34474][formats] Fix spotless format violation in test Spotless requires the assertThatThrownBy chain on a single line. --- .../flink/formats/avro/AvroDeserializationSchemaTest.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroDeserializationSchemaTest.java b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroDeserializationSchemaTest.java index 456c4a5594f675..454b22dea0447d 100644 --- a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroDeserializationSchemaTest.java +++ b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroDeserializationSchemaTest.java @@ -118,8 +118,7 @@ void testDeserializeRecoversFromCorruptMessage(AvroEncoding encoding) throws Exc byte[] corrupt = new byte[] {100, 0, 0, 0, 0, 0, 0, 0}; assertThat(deserializer.deserialize(validBytes)).isEqualTo(valid); - assertThatThrownBy(() -> deserializer.deserialize(corrupt)) - .isInstanceOf(Exception.class); + assertThatThrownBy(() -> deserializer.deserialize(corrupt)).isInstanceOf(Exception.class); // FLINK-34474: a subsequent valid message must still deserialize, // instead of being poisoned by the prior failed read. assertThat(deserializer.deserialize(validBytes)).isEqualTo(valid); From 412fa61554eb8180a8a3c10899f1c152a7a0d337 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=83=A1=E7=90=B3=28HuLin=29?= Date: Thu, 6 Aug 2026 15:40:55 +0800 Subject: [PATCH 3/3] [FLINK-34474][formats] Fix binaryDecoder type mismatch in resetDecoder DecoderFactory.binaryDecoder(InputStream, BinaryDecoder) requires a BinaryDecoder reuse arg, but this.decoder is declared as Decoder (it may also hold a JsonDecoder). Pass null instead to discard the poisoned decoder and build a fresh BinaryDecoder. --- .../apache/flink/formats/avro/AvroDeserializationSchema.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroDeserializationSchema.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroDeserializationSchema.java index e07a0f3cbe9b81..a60014571ae582 100644 --- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroDeserializationSchema.java +++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroDeserializationSchema.java @@ -198,7 +198,10 @@ private void resetDecoder() { if (encoding == AvroEncoding.JSON) { this.decoder = DecoderFactory.get().jsonDecoder(getReaderSchema(), inputStream); } else { - this.decoder = DecoderFactory.get().binaryDecoder(inputStream, this.decoder); + // Rebuild the BinaryDecoder bound to the input stream. The pooled + // decoder is discarded (passing null as reuse) so a poisoned internal + // buffer cannot leak into the next message. + this.decoder = DecoderFactory.get().binaryDecoder(inputStream, null); } } catch (IOException e) { // jsonDecoder only throws on schema/input issues that cannot occur here