kafka-avro-console-producer not respecting --property parse.headers=true

[aymkhalil]

I'm using the confluentinc/cp-schema-registry:7.2.1 docker image and trying to produce Avro messages with headers. The command is:

printf 'h1:v1,h2:v2\t{"field":"value0"}' | kafka-avro-console-producer --broker-list broker:9092 --property 'value.schema={"type":"record","name":"test","namespace":"ns","fields":[{"name":"field","type":"string"}]}' --property schema.registry.url=http://localhost:8081 --property parse.headers=true --topic users

The exception I'm getting (seems the reader is not respecting the \t delimiter and trying to parse the entire input as JSON):

[2022-09-23 05:03:00,996] INFO Kafka startTimeMs: 1663909380987 (org.apache.kafka.common.utils.AppInfoParser)
org.apache.kafka.common.errors.SerializationException: Error deserializing json h1:v1,h2:v2	{"field":"value0"} to Avro of schema {"type":"record","name":"test","namespace":"ns","fields":[{"name":"field","type":"string"}]}
	at io.confluent.kafka.formatter.AvroMessageReader.readFrom(AvroMessageReader.java:131)
	at io.confluent.kafka.formatter.SchemaMessageReader.readMessage(SchemaMessageReader.java:325)
	at kafka.tools.ConsoleProducer$.main(ConsoleProducer.scala:50)
	at kafka.tools.ConsoleProducer.main(ConsoleProducer.scala)
Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'h1': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
 at [Source: (String)"h1:v1,h2:v2\u0009{"field":"value0"}"; line: 1, column: 3]
	at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:2391)
	at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:745)
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._reportInvalidToken(ReaderBasedJsonParser.java:2961)
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._handleOddValue(ReaderBasedJsonParser.java:2002)
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:802)
	at org.apache.avro.io.JsonDecoder.configure(JsonDecoder.java:124)
	at org.apache.avro.io.JsonDecoder.<init>(JsonDecoder.java:66)
	at org.apache.avro.io.JsonDecoder.<init>(JsonDecoder.java:74)
	at org.apache.avro.io.DecoderFactory.jsonDecoder(DecoderFactory.java:266)
	at io.confluent.kafka.schemaregistry.avro.AvroSchemaUtils.toObject(AvroSchemaUtils.java:215)
	at io.confluent.kafka.formatter.AvroMessageReader.readFrom(AvroMessageReader.java:124)
	... 3 more

Running the same command (leaving out the schema related config) with the kafka-console-producer tool works just fine:

printf 'h1:v1,h2:v2\t{"field":"value0"}' | kafka-console-producer --broker-list broker:9092 --property parse.headers=true --topic users

[OneCricketeer]

Related (opposite for consumer) #1927

Similarly, headers aren't parsed. https://github.com/confluentinc/schema-registry/blob/master/schema-serializer/src/main/java/io/confluent/kafka/formatter/SchemaMessageReader.java#L318

[OneCricketeer]

As a workaround, kcat supports headers (and Avro)

https://github.com/edenhill/kcat#examples

[DennisFederico]

Doesn't seem to be fixed on 7.3.3

[rayokota]

@DennisFederico , this was added to 7.4.0+

[DennisFederico]

OK