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

flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroDeserializationSchema.java

@@ -181,7 +181,33 @@ 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 {
+                // 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
+            // (the schema is cached and the input is an in-memory stream).
+            throw new RuntimeException(e);
+        }
     }
 
     void checkAvroInitialized() throws IOException {

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,38 @@ 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<GenericRecord> 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);
+    }
 }