Make AvroDeserializationSchema deser failure tolerant

Author: ferenc-csaky

flink-sql-runner/pom.xml

@@ -233,6 +233,13 @@
                 </excludes>
               </artifactSet>
               <filters>
+                <!-- Exclude original AvroDeserializationSchema, because for now we overwrite that class downstream. -->
+                <filter>
+                  <artifact>org.apache.flink:flink-avro</artifact>
+                  <excludes>
+                    <exclude>org/apache/flink/formats/avro/AvroDeserializationSchema**</exclude>
+                  </excludes>
+                </filter>
                 <filter>
                   <artifact>*:*</artifact>
                   <excludes>

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

@@ -173,7 +173,16 @@ 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 (Exception e) {
+      /*
+       * If the deserialization fails, {@code datumReader} has to be reset, because {@code Decoder} will remain in an
+       * inconsistent state, which will make subsequent valid deserialization attempts to fail.
+       */
+      this.datumReader = null;
+      throw e;
+    }
   }
 
   void checkAvroInitialized() throws IOException {