Problem

You have different versions in crates.io vs Github. Crates.io seems like a later version. So probably need to update the Github version.

The crates.io version has a problem in this code:

    pub async fn send_message(&self, topic_name: &str, message: Message) {
        let json_payload = to_string(&message.value).expect("Failed to serialize message");

        let delivery_status = self
            .producer
            .send(
                FutureRecord::to(topic_name)
                    .payload(&json_payload)
                    .key(&message.key),
                Duration::from_secs(0),
            )
            .await;

        // Log the status of the message delivery.
        match delivery_status {
            Ok((partition, offset)) => {
                info!(
                    "Message with key {} delivered to partition {} at offset {}",
                    message.key, partition, offset
                );
            }
            Err((error, _message)) => {
                error!(
                    "Failed to deliver message with key {}: {}",
                    message.key, error
                );
            }
        }
    }

I'm trying to send Line Protocol string to Kafka so that Kafka can send it to InfluxDB. I insert the value exactly as it should be in the request body, into the value field of Message, which I insert into message argument in send_message. But the code changes this string value into something else (by adding backslashes and quotes for example) in this line:

let json_payload = to_string(&message.value).expect("Failed to serialize message");

When this altered string value goes to Kafka, then Kafka is unable to send it to InfluxDB because the string value has changed so that it's no longer valid Line Protocol syntax.

I tried to send PR but since the Github version is old relative to the Crates.io version, I'll instead add the requested change below:

Solution

Replace this:

        let json_payload = to_string(&message.value).expect("Failed to serialize message");

...with this:

let json_payload = message.value;

cploutarchou / tiny_kafka

`tiny_kafka` ruins the value before sending to Kafka server #2

Problem

Solution