How to use use EasyProtoRawEncoder?

[hongbo-miao]

I am wondering how to use schema_registry_converter::async_impl::easy_proto_raw::EasyProtoRawEncoder.
We have huge data streaming, and Protobuf will be faster than AVRO.

I searched across all GitHub repos, but no one is using it currently: https://sourcegraph.com/search?q=context:global+EasyProtoRawEncoder&patternType=keyword&sm=0

Also, I am not sure how to add this dependency? I tried

schema_registry_converter = { version = "4.2.0", features = ["proto", "easy"] }

but seems proto does not exist.

Here is my best try I made

use chrono::Utc;
use rand::Rng;
use rdkafka::producer::{FutureProducer, FutureRecord};
use rdkafka::util::Timeout;
use rdkafka::ClientConfig;
use schema_registry_converter::async_impl::easy_proto_raw::EasyProtoRawEncoder;
use schema_registry_converter::async_impl::schema_registry::SrSettings;
use schema_registry_converter::schema_registry_common::SubjectNameStrategy;
use serde::Serialize;
use std::env::args;
use std::time::Duration;
use tokio::time;

#[derive(Serialize)]
struct SensorData {
    motor_id: String,
    timestamp: String,
    temperature1: f64,
    temperature2: f64,
}

fn create_producer(bootstrap_server: &str) -> FutureProducer {
    ClientConfig::new()
        .set("bootstrap.servers", bootstrap_server)
        .create()
        .expect("Failed to create producer")
}

fn generate_sensor_data(motor_id: &str) -> SensorData {
    let mut rng = rand::thread_rng();
    let temperature = rng.gen_range(10.0..100.0);
    SensorData {
        motor_id: motor_id.to_string(),
        timestamp: Utc::now().to_rfc3339(),
        temperature1: temperature,
        temperature2: temperature,
    }
}

#[tokio::main]
async fn main() {
    println!("Starting IoT Data Generator...");

    // Get bootstrap server from args or use default
    let bootstrap_server = args()
        .nth(1)
        .unwrap_or_else(|| "localhost:9092".to_string());
    let producer = create_producer(&bootstrap_server);

    // Initialize
    let schema_registry_url = "https://confluent-schema-registry.hongbomiao.com";
    let sr_settings = SrSettings::new(schema_registry_url.to_string());
    let encoder = EasyProtoRawEncoder::new(sr_settings);

    // Simulate 3 different IoT motors
    let motor_ids = vec!["motor_001", "motor_002", "motor_003"];
    let topic = "production.iot.motor.proto";

    println!("Sending data to Kafka topic: {}", topic);

    // Create an interval for sending data
    let mut interval = time::interval(Duration::from_nanos(1000000));

    loop {
        interval.tick().await;

        for motor_id in &motor_ids {
            let sensor_data = generate_sensor_data(motor_id);

            // Convert data to Avro bytes
            let avro_payload = encoder
                .encode_struct(
                    &sensor_data,
                    &SubjectNameStrategy::TopicNameStrategy(topic.to_string(), false),
                )
                .await
                .expect("Failed to encode sensor data to Avro");

            match producer
                .send(
                    FutureRecord::to(topic)
                        .payload(&avro_payload)
                        .key(motor_id.as_bytes()),
                    Timeout::After(Duration::from_secs(1)),
                )
                .await
            {
                Ok((partition, offset)) => {
                    println!(
                        "Sent data for motor {} to partition {} at offset {}",
                        motor_id, partition, offset
                    );
                }
                Err((err, _)) => {
                    eprintln!("Failed to send data for motor {}: {}", motor_id, err);
                }
            }
        }
    }
}

My schema looks like

syntax = "proto3";
package com.hongbomiao;

import "production.iot.motor.proto";

message SilData {
  optional string motor_id = 1;
  optional string timestamp = 2;
  optional double temperature1 = 3;
  optional double temperature2 = 4;
}

It would be great to add an demo, thanks! ☺️

[gklijs]

There is a test already. Adding demo's for all possible options will take a lot of time.

[hongbo-miao]

Thank you so much @gklijs ! I didn't notice the test is inside source code.
I think I am very close.
Only part I am not super clear is

schema_registry_converter/src/async_impl/easy_proto_raw.rs

Line 96 in 04fb737

.with_body(get_proto_body(get_proto_hb_schema(), 7))

which calls

schema_registry_converter/test_utils/src/lib.rs

Lines 29 to 35 in 04fb737

	pub fn get_proto_hb_101_only_data() -> &'static [u8] {
	&get_proto_hb_101()[6..]
	}
	pub fn get_proto_hb_101() -> &'static [u8] {
	&[0, 0, 0, 0, 7, 0, 8, 101]
	}

One thing I noticed current readme didn't mention feature proto_raw (it mentioned easy, avro, maybe worth mentioning.

I will build a consumer to see if I can consumer properly, and will post a full working solution here once I confirmed it works. Thanks! ☺️

[hongbo-miao]

Thank you @gklijs for the guide! I just succeed for both producer and consumer. Posted my solution at #122