Consuming a key/value from KAFKA using kafka-json-schema-console-consumer only returns value but not key
I am building a Kafka source connector and am able to successfully publish a Kafka SourceRecord with a key and its schema with a value and its schema. However, when I use kafka-json-schema-console-consumer to consume the message on the topic I expect both key and value to be received, but I only receive the value and the key is NOT received. Any help is appreciated!
Below is the relevant code that I have tried/used:
The message that I am publishing:
"message": {
"key": {
"id": "some-id"
},
"data": {
"text": "some-text"
}
}
The message gets put into he SourceRecord as follows:
:
:
SourceRecord sr = new SourceRecord(
sourcePartition, sourceOffset, this.topicName, partition,
keySchema, key, valueSchema, value);
log.info("HttpSourceTask:buildSourceRecord - source:{}", sr);
I use a log message to confirm the message was successfully processed… below is the console log message in Kafka which confirms that the key plus key schema and the value plus value schema are correctly set in the SourceRecord:
connect | [2022-01-29 21:24:47,455] INFO HttpSourceTask:buildSourceRecord - source:SourceRecord{sourcePartition={source=bgs-topic}, sourceOffset={offset=0}} ConnectRecord{topic="bgs-topic", kafkaPartition=null, key=Struct{id=some-id}, keySchema=Schema{STRUCT}, value=Struct{text=some-text}, valueSchema=Schema{STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
Note that the key and keySchema fields are filled in with a STRUCT (as are the value and valueSchema fields).
I am using the following code (I am using Kafka in docker) to consume the message… this is the part that DOES NOT seem to return the right data (key is missing).
docker exec -it schema-registry
/usr/bin/kafka-json-schema-console-consumer
--bootstrap-server http://kafka:9092
--topic bgs-topic
--from-beginning
--property key.separator="|"
--property value.schema="
{
"title": "simple-data-schema",
"type" : "object",
"required" : [ "text" ],
"properties" : {
"text" : {
"type": "string"
}
}
}"
--property parse.key=true
--property key.schema="
{
"title": "simple-key-schema",
"type" : "object",
"required" : [ "id" ],
"properties" : {
"id" : {
"type": "string"
}
}
}"
Here is the returned message – note that the only value is available:
{"text":"some-text"}
Read more here: Source link