How can i send key : value to kafka with kafkaconsoleproducer without it beind encoded to json?-CodePudding

I'm sending a message to kafka manually using kafka-console-producer where key and value are jsons. When it arrives it seems to be encoded to json again and i don't know how to construct the message to be encoded to json i want to send or how to configure either the producer I'm using or the kafka broker itself not to encode messages.

Here is example:

I have file message.json:

{"distributionID": "TB-AARTool-30","senderID": "TB-AARTool","dateTimeSent": 1635405698193,"dateTimeExpires": 1635405698193,"distributionStatus": "System","distributionKind": "Request"};{"TrialId": 3,"TrialSessionId": 5,"TrialStageId": 4}

I'm sending it using command:

./kafka-console-producer.sh --broker-list localhost:9092 --topic topic --property "parse.key=true" --property "key.separator=;" < /message.json

and when it arrives it looks like:

 {
    "topic": "topic",
    "key": "{\"distributionID\": \"TB-AARTool-30\",\"senderID\": \"TB-AARTool\",\"dateTimeSent\": 1635405698193,\"dateTimeExpires\": 1635405698193,\"distributionStatus\": \"System\",\"distributionKind\": \"Request\"}",
    "value": "{\"TrialId\": 3,\"TrialSessionId\": 5,\"TrialStageId\": 4}",
    "partition": 0,
    "offset": 1
  }

First of all at the beginning and end of the key and value quotation marks (") were added. Second, all quotation marks (") already in the message were replaced with backslash quotation mark (\"). I just want key and value to be received by kafka as valid jsons.

I'm using:

confluentinc/cp-kafka:latest docker image which i presume is latest stable kafka server version.
kafka_2.12-3.0.0 from which I'm using bin/kafka-console-producer.sh (https://www.apache.org/dyn/closer.cgi?path=/kafka/3.0.0/kafka_2.12-3.0.0.tgz) which also contains kafka server and zookeepr but i don't use them.

CodePudding user response：

Your data produced fine. Try using kafka-console-consumer instead that will not show escaped JSON strings.

Regarding the shown output, appears as if you are consuming your record from some REST API or other non-Kafka built-in tool? In that case, yes, your key and value are escaped strings because that is what was produced - strings, not JSON objects.

Therefore, you aren't able to get data like this without customizing whatever deserializer gets this output

{
    "topic": "topic",
    "key": {
       "distributionID": ...
        ...
    }
    "value": { 
        "TrialId":  ...
        ...   
    } 
    ...
}

This isn't a problem for other consumers that are configured to use StringDeserializer as you can pass those strings into a JSON parser to get objects