Consuming messages causes json.decoder.JSONDecodeError

[Glutexo]

When consuming a message set from Kafka, a JSONDecodeError is raised. This is caused by decoding a message with offset -1 and value '' (empty string), which is not valid JSON. Whenever a message is consumed by BaseConsumer, after successfully processing the actual message, another invalid one appears in the message set.

I created a gist with a minimal Tornado app. It connects to a single Kafka topic, reads a JSON message from stdin, produces it to Kafka and consumes it from there. Every time the consuming tries to parse two messages: first the one produced, which succeeds, and second some ghost-message with negative offset and empty value.

I am running with following versions:

• tornado 5.1
• kiel 0.9.3
• confluentinc/cp-kafka Docker image with Kafka version 2.0.0

[theoren]

Suggested workaround fix - see fork https://github.com/theoren/kiel/commits/master/