Logstash 5.4 - Repeated deserialization errors and losing data with Persistent Queues

The error stack trace is very similar to: Logstash Crash with Persistent Queue and Kafka Input except that issue uses Kafka, we use PQs.