Add support for reading Consumer mode JSON back into Producer

[AlexeiZenin]

Currently I'm facing an issue where the Kafka key is JSON and I chose the ":" delimiter by default without thinking about it too much (the values are also JSON). This lead to keys and values being messed up when I produced back to Kafka through kafkacat.

Kafkacat file produced by consuming from topic with : delimiter (kafkacat -C -e -J -q -b localhost:9092 -t _schemas > schemas.log):

{"keytype":"NOOP","magic":0}:
{"keytype":"NOOP","magic":0}:
{"keytype":"NOOP","magic":0}:
{"keytype":"SCHEMA","subject":"protoexample","version":1,"magic":1}:{"subject":"protoexample","version":1,"id":1,"schemaType":"PROTOBUF","schema":"syntax = \"proto3\";\n\nmessage Test {}\n","deleted":false}
{"keytype":"SCHEMA","subject":"protoexample","version":2,"magic":1}:{"subject":"protoexample","version":2,"id":2,"schemaType":"PROTOBUF","schema":"syntax = \"proto3\";\n\nmessage Test2 {}\n","deleted":false}
{"keytype":"SCHEMA","subject":"protoexample","version":3,"magic":1}:{"subject":"protoexample","version":3,"id":3,"schemaType":"PROTOBUF","schema":"syntax = \"proto3\";\n\nmessage Test3 {}\n","deleted":false}

I could choose some magical character that probably has a high likelihood of never appearing in my data, but to be 100% foolproof I think kafkacat can add support to consume its own JSON format structure back into its Producer mode.

It would be much more robust and allow users to just be able to quickly save the JSON from the Consumer mode to a file and then read from it without needing to think about delimiters at all and possible collisions with actual characters within their data (either the key or the value).

Given that the JSON is well formed and organized it should be possible to expose an extra -J flag on the Producer side of things to support parsing the -J output produced in the Consumer mode.

JSON mode for Consumer mode:

{"topic":"_schemas","partition":0,"offset":0,"tstype":"create","ts":1614201077742,"broker":1,"key":"{\"keytype\":\"NOOP\",\"magic\":0}","payload":null}
{"topic":"_schemas","partition":0,"offset":1,"tstype":"create","ts":1614201078370,"broker":1,"key":"{\"keytype\":\"NOOP\",\"magic\":0}","payload":null}
{"topic":"_schemas","partition":0,"offset":2,"tstype":"create","ts":1614221939056,"broker":1,"key":"{\"keytype\":\"NOOP\",\"magic\":0}","payload":null}
{"topic":"_schemas","partition":0,"offset":3,"tstype":"create","ts":1614266108892,"broker":1,"key":"{\"keytype\":\"SCHEMA\",\"subject\":\"protoexample\",\"version\":1,\"magic\":1}","payload":"{\"subject\":\"protoexample\",\"version\":1,\"id\":1,\"schemaType\":\"PROTOBUF\",\"schema\":\"syntax = \\\"proto3\\\";\\n\\nmessage Test {}\\n\",\"deleted\":false}"}
{"topic":"_schemas","partition":0,"offset":4,"tstype":"create","ts":1614266197654,"broker":1,"key":"{\"keytype\":\"SCHEMA\",\"subject\":\"protoexample\",\"version\":2,\"magic\":1}","payload":"{\"subject\":\"protoexample\",\"version\":2,\"id\":2,\"schemaType\":\"PROTOBUF\",\"schema\":\"syntax = \\\"proto3\\\";\\n\\nmessage Test2 {}\\n\",\"deleted\":false}"}
{"topic":"_schemas","partition":0,"offset":5,"tstype":"create","ts":1614266200301,"broker":1,"key":"{\"keytype\":\"SCHEMA\",\"subject\":\"protoexample\",\"version\":3,\"magic\":1}","payload":"{\"subject\":\"protoexample\",\"version\":3,\"id\":3,\"schemaType\":\"PROTOBUF\",\"schema\":\"syntax = \\\"proto3\\\";\\n\\nmessage Test3 {}\\n\",\"deleted\":false}"}

This way, as seen above, the producer would not need delimiters at all and just parse a well formed JSON object to extract the key and payload values.

[masoncj]

This has been implemented over in #295. Check it out!