[FLINK-34474] After failed deserialization with ConfluentRegistryAvroDeserializationSchema all subsequent deserialization fails

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 1.18.1
• Component/s: Formats (JSON, Avro, Parquet, ORC, SequenceFile)

Environment

• Locally executed, without Flink cluster.
• Flink on Kubernetes

• Assignee: Unassigned
• Reporter: Grzegorz Liter
• Created: 20/Feb/24 13:47
• Updated: 20/Feb/24 15:39

Description

Steps to reproduce:

1. Create ConfluentRegistryAvroDeserializationSchema instance for specific Avro
2. Parse invalid byte representation of serialized Avro event
3. Parse valid byte representation of serialized Avro event

Expected:
Validation in step 3 is successful

 

Actual:
Validation in step 3 fails

Short code example, I cannot attach full examples at this time:

```
public class DeserializationTest {
public static void main(String[] args) throws Exception {
byte[] valid = new byte[]

{ ... }

;
byte[] invalid = new byte[]

{ ... }

;

ConfluentRegistryAvroDeserializationSchema<RawEvent> deserializer = ConfluentRegistryAvroDeserializationSchema.forSpecific(RawEvent.class, valid schema registry url);

System.out.println("deserialize valid");
des(deserializer, valid);
System.out.println("deserialize invalid");
des(deserializer, invalid);
System.out.println("deserialize valid");
des(deserializer, valid);
System.out.println("deserialize valid");
des(deserializer, valid);
}

private static void des(ConfluentRegistryAvroDeserializationSchema<RawEvent> deserializer, byte[] bytes) {
try

{ deserializer.deserialize(bytes); System.out.println("VALID"); }

catch (Exception e)

{ System.out.println("FAILED: " + e); }

}
}
```

Console output:

```
deserialize valid
VALID
deserialize invalid
FAILED: java.lang.ArrayIndexOutOfBoundsException: Index -154 out of bounds for length 2
deserialize valid
FAILED: java.lang.ArrayIndexOutOfBoundsException: Index 24 out of bounds for length 2
deserialize valid
FAILED: java.lang.ArrayIndexOutOfBoundsException: Index 25 out of bounds for length 2
```