Continue pipeline after invalid data

[hatamiarash7]

I have a Kafka source with a huge amount of JSON data. This data was not validated before production and had many problems, such as non-UTF8 characters, bad integer data ( like "key": 004 ), corrupted JSON, etc.

In the source configuration, I choose the Drop option to ignore bad data. But this will not work, and the processing is stopped after the first bad data.

As you can see in the screenshot, I have a stdout sink to see the output of the query. Everything works fine, but immediately after the first bad data, I get a warning, but the processing stops, and nothing is sent to the sink.

[kcharkseliani]

Investigated this - the issue here is that when an invalid JSON messages comes in, the tape decoder errors on reaching an invalid symbol, but then the previous part of the message stays in the buffer (shortcoming of the tape decoder), preventing further valid messages from getting processed. This also means that flushing will never work after that point due to truncated JSON error (again, a shortcoming of the tape decoder that needs all messages to be complete before flushing)

For example, when something like {"timestamp": 1234567890iiii, "width": 100} comes in, the decoder errors on reaching symbol 'i' with error:
WARN arroyo_operator::context: Dropping invalid data: invalid JSON: JsonError("Encountered unexpected 'i' whilst parsing object")
After that, {"timestamp": 1234567890 stays in the buffer, so when a valid message like {"timestamp": 1234567891, "width": 120} comes in, we get:
WARN arroyo_operator::context: Dropping invalid data: invalid JSON: JsonError("Encountered unexpected '{' whilst parsing object")

I made a fix for this that also keeps track of all incomplete messages in the buffer so that JSON messages sent across in chunks can be buffered and flushed properly, with invalid messages not sent to the decoder at all - #891 JSON buffering fix when bad data drop is enabled

@mwylde as I'm not a contributor for the repo, I created this PR from my fork. Would you be able to take a look please?

[hatamiarash7]

Thank you @kcharkseliani