Kafka reader deadloop when buffer has reached max capacity

[rixed]

After the failure the buffer is not reset, thus the next line is likely to triger the exact same error, and now all lines are going to trigger the error.