Confluent Cloud client example: Rust #170

frankgreco · 2019-06-05T05:07:48Z

@confluentinc/devx

ybyzek · 2019-06-05T11:02:06Z

@frankgreco can you please add Rust to https://github.com/confluentinc/examples/blob/5.2.1-post/clients/cloud/README.md ?

ybyzek · 2019-06-05T11:04:14Z

clients/cloud/rust/README.md

+
+```bash
+$ RUSTFLAGS='-C link-args=-lzstd' cargo build
+$ ./target/debug/producer --config ~/.ccloud/example.config --topic i_love_rust


For consistency with all the other client examples, let's have topic name test1

ybyzek · 2019-06-05T11:06:57Z

clients/cloud/rust/README.md

+Produce messages to produce and consume messages from [Confluent Cloud](https://www.confluent.io/confluent-cloud/) using the [rust-rdkafka client for Apache Kafka](https://github.com/fede1024/rust-rdkafka).
+
+# Prerequisites
+* [Rust Client for Apache Kafka](https://github.com/fede1024/rust-rdkafka#installation) installed on your machine


Client -> client

ybyzek · 2019-06-05T11:10:42Z

clients/cloud/rust/src/producer.rs

+fn log_produce_result(result: Result<(i32, i64), (KafkaError, OwnedMessage)>) -> Result<(), ()> {
+    result
+        .and_then(|(p, o)| {
+            println!("produced to partition {} with offset {}", p, o);


For consistency with the other client examples, can this print topic name as well? Specifically, this is what the other clients print:

Produced record to topic test1 partition [0] @ offset 0 ...

ybyzek · 2019-06-05T11:12:20Z

clients/cloud/rust/src/producer.rs

+
+    let messages = (0..9)
+        .map(|msg| {
+            producer.send(


For consistency with the other client examples, can this print that it's about to send? Specifically, this is what the other clients print:

Producing record: alice {"count": 0} ...

ybyzek · 2019-06-05T15:53:22Z

clients/cloud/README.md

@@ -22,7 +22,7 @@ The following subset includes examples with Confluent Cloud Schema Registry and
 | [![](images/kafka-connect-datagen.png)](kafka-connect-datagen/) | [ksql-datagen](ksql-datagen/) | |


-# Other Confluent Cloud Demos
+# Other Confluent Cloud Dem


@frankgreco -- typo

edenhill · 2019-06-05T19:59:18Z

clients/cloud/README.md

@@ -10,7 +10,7 @@ This directory includes examples of Kafka client applications connecting to [Con
 | [![](images/scala.png)](scala/)   | [![](images/confluent-cli.png)](confluent-cli/) | [![](images/ruby.png)](ruby/)     |
 | [![](images/groovy.png)](groovy/) | [![](images/kotlin.png)](kotlin/)               | [![](images/nodejs.png)](nodejs/) |
 | [![](images/kafkacat.jpg)](kafkacat/) | [![](images/dotnet.png)](csharp/)           | [![](images/c.png)](c/) |
-| [![](images/kafka-connect-datagen.png)](kafka-connect-datagen/) | [ksql-datagen](ksql-datagen/) | |
+| [![](images/kafka-connect-datagen.png)](kafka-connect-datagen/) | [ksql-datagen](ksql-datagen/) | [<img src="images/rust.png" width="100" />](rust/) |


This image import should probably not be different from the existing ones.

the reason it is is because this is the only way in GH markdown that I know of to resize an image.

Okay, can we resize the actual image instead? We did this with the other ones

yea sure I could do that - 256x256 seems like it should fit

edenhill · 2019-06-05T20:00:25Z

clients/cloud/rust/.gitignore

@@ -0,0 +1,2 @@
+/target


This should be just target, right? /target is outside the git repo

this .gitignore was generated by the cargo cli

edenhill · 2019-06-05T20:01:25Z

clients/cloud/rust/README.md

@@ -0,0 +1,61 @@
+# Overview
+
+Produce messages to produce and consume messages from [Confluent Cloud](https://www.confluent.io/confluent-cloud/) using the [rust-rdkafka client for Apache Kafka](https://github.com/fede1024/rust-rdkafka).


"Produce messages to produce .." is redundant, should probably be "Produce messages to and consume messages from .."

copy/paste from other client examples - I left it as is for uniformity

edenhill · 2019-06-05T20:02:13Z

clients/cloud/rust/README.md

+```
+
+# Example 1: Hello World!
+In this example, the producer writes Kafka data to a topic in Confluent Cloud. 


"writes data to a Kafka topic"

copy/paste from other client examples - I left it as is for uniformity

I guess I could change in all client example readme's

edenhill · 2019-06-05T20:04:04Z

clients/cloud/rust/README.md

+1. Run the producer, passing in arguments for (a) the local file with configuration parameters to connect to your Confluent Cloud instance and (b) the topic name:
+
+```bash
+$ RUSTFLAGS='-C link-args=-lzstd' cargo build


Why are these RUSTFLAGS needed? librdkafka should already be linked with -lzstd

I would love to figure out how to do a build without specifying this but I spend too much time trying to get it to work and this was the workaround. Here is the error I get:

= note: Undefined symbols for architecture x86_64: "_ZSTD_createCStream", referenced from: _rd_kafka_zstd_compress in librdkafka_sys-347d93b86e02028e.rlib(rdkafka_zstd.o) "_ZSTD_initCStream", referenced from: _rd_kafka_zstd_compress in librdkafka_sys-347d93b86e02028e.rlib(rdkafka_zstd.o) "_ZSTD_endStream", referenced from: _rd_kafka_zstd_compress in librdkafka_sys-347d93b86e02028e.rlib(rdkafka_zstd.o) "_ZSTD_decompress", referenced from: _rd_kafka_zstd_decompress in librdkafka_sys-347d93b86e02028e.rlib(rdkafka_zstd.o) "_ZSTD_compressBound", referenced from: _rd_kafka_zstd_compress in librdkafka_sys-347d93b86e02028e.rlib(rdkafka_zstd.o) "_ZSTD_isError", referenced from: _rd_kafka_zstd_decompress in librdkafka_sys-347d93b86e02028e.rlib(rdkafka_zstd.o) _rd_kafka_zstd_compress in librdkafka_sys-347d93b86e02028e.rlib(rdkafka_zstd.o) "_ZSTD_compressStream", referenced from: _rd_kafka_zstd_compress in librdkafka_sys-347d93b86e02028e.rlib(rdkafka_zstd.o) "_ZSTD_freeCStream", referenced from: _rd_kafka_zstd_compress in librdkafka_sys-347d93b86e02028e.rlib(rdkafka_zstd.o) "_ZSTD_getFrameContentSize", referenced from: _rd_kafka_zstd_decompress in librdkafka_sys-347d93b86e02028e.rlib(rdkafka_zstd.o) "_ZSTD_getErrorCode", referenced from: _rd_kafka_zstd_decompress in librdkafka_sys-347d93b86e02028e.rlib(rdkafka_zstd.o) "_ZSTD_getErrorName", referenced from: _rd_kafka_zstd_decompress in librdkafka_sys-347d93b86e02028e.rlib(rdkafka_zstd.o) _rd_kafka_zstd_compress in librdkafka_sys-347d93b86e02028e.rlib(rdkafka_zstd.o) ld: symbol(s) not found for architecture x86_64 clang: error: linker command failed with exit code 1 (use -v to see invocation)

This looks like a bug, it should be possible to automate this with a build.rs file
fede1024/rust-rdkafka#125

@gardnervickers good find! let me try that 💯

edenhill · 2019-06-05T20:11:36Z

clients/cloud/rust/src/utils.rs

@@ -0,0 +1,53 @@
+extern crate clap;


copyright blobbing

edenhill · 2019-06-05T20:12:17Z

clients/cloud/rust/src/utils.rs

+        .get_matches();
+
+    let mut kafka_config = ClientConfig::new();
+    kafka_config.set_log_level(RDKafkaLogLevel::Debug);


Not needed. No debug is logged unless you specify a "debug" config property, in which case it also raises the log level.

edenhill · 2019-06-05T20:12:51Z

clients/cloud/rust/src/utils.rs

+    let file = File::open(matches.value_of("config").ok_or("error parsing config")?)?;
+    for line in BufReader::new(&file).lines() {
+        let cur_line = line?;
+        if cur_line.starts_with('#') || cur_line.chars().count() < 1 {


Might want to l/r trim the line.

edenhill · 2019-06-05T20:13:16Z

clients/cloud/rust/src/utils.rs

+            continue;
+        }
+        let key_value: Vec<&str> = cur_line.split("=").collect();
+        kafka_config.set(


will this propagate errors?

anywhere where there is an error, a result is returned and errors are propagated

edenhill · 2019-06-05T20:14:06Z

clients/cloud/rust/src/consumer.rs

+
+fn main() -> Result<(), Box<std::error::Error>> {
+    let (topic, mut config) = utils::get_config()?;
+    let consumer: StreamConsumer = config.set("group.id", "rust_example_group_1").create()?;


it needs some more config for cloud, such as "security.protocol" and "sasl.mechanisms". Same for the producer

those are provided by the passed in config

But the config file example above does not have them.

good point - i'll make it clear in the readme that they should be there by adding them to the example

gardnervickers · 2019-06-05T15:21:11Z

clients/cloud/rust/src/consumer.rs

+fn echo_message<M: Message>(msg: M) -> Result<(), std::str::Utf8Error> {
+    let deserialize = |o| match o {
+        None => Ok("".to_string()),
+        Some(val) => Ok(std::str::from_utf8(val)?.to_string()),


I think this can just be String::from_utf8(val)

If I do that, then I need to use .to_vec() which then means I need add type annotation to the closure. Hence, it's less code to have it as is.

Hmm on second thought, we're taking a &[u8] here, converting it to a &str and then to a String. I think we should be able to avoid the to_string() copying the &str to a String here right?

ah good observation - no need to convert to a String

gardnervickers · 2019-06-05T15:23:02Z

clients/cloud/rust/src/consumer.rs

+                None
+            }
+        })
+        .for_each(move |msg| {


Hmm do we need the move here?

nope :) good catch - I wish the compiler told you this because at one point I probably did but then when I no longer needed it it's still valid code and doesn't warn.

gardnervickers · 2019-06-05T21:36:28Z

clients/cloud/rust/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "rust_kafka_client_example"


I think the convention is to use dashes instead of underscores here

gardnervickers · 2019-06-05T21:37:51Z

clients/cloud/rust/README.md

+1. Run the producer, passing in arguments for (a) the local file with configuration parameters to connect to your Confluent Cloud instance and (b) the topic name:
+
+```bash
+$ RUSTFLAGS='-C link-args=-lzstd' cargo build


This looks like a bug, it should be possible to automate this with a build.rs file
fede1024/rust-rdkafka#125

gardnervickers · 2019-06-05T21:39:03Z

clients/cloud/rust/src/consumer.rs

+ */
+extern crate futures;
+extern crate rdkafka;
+extern crate tokio;


It might be worth considering using Rust 2018 and removing these extern crate statements.

Nice! Didn't know these weren't required in Rust 2018 :)

gardnervickers · 2019-06-05T21:39:52Z

clients/cloud/rust/src/consumer.rs

+    Ok(())
+}
+
+fn main() -> Result<(), Box<std::error::Error>> {


Should this be dyn std::error::Error?

I got the inspiration from here.

gardnervickers · 2019-06-05T21:42:20Z

clients/cloud/rust/src/producer.rs

+mod utils;
+
+fn log_produce_result(
+    topic: String,


Does topic here need to be owned or can it be &str?

You are right, it doesn't need to be owned!

gardnervickers · 2019-06-05T21:42:59Z

clients/cloud/rust/src/producer.rs

+    for msg in messages {
+        msg.wait()
+            .map_err(|err| eprintln!("error producing message: {}", err))
+            .and_then(|result| log_produce_result(topic.to_owned(), result))


Im not sure topic needs to be cloned here if log_produce_result does not need an owned String.

frankgreco · 2019-06-06T15:20:21Z

@edenhill @ybyzek @gardnervickers I believe I have addressed all your suggestions 😄

ybyzek

@frankgreco thanks for the PR! Can you resize the Rust graphic just a tough smaller (shave off ~15%), but otherwise LGTM! I approve.

edenhill

LGTM!

edenhill · 2019-06-07T17:23:31Z

clients/cloud/rust/Cargo.toml

+name    = "rust-kafka-client-example"
+version = "0.1.0"
+authors = ["GRECO, FRANK <[email protected]>"]
+edition = "2018"


that's not the current year, but the rust edition

gardnervickers

LGTM, one thing to watch out for is that trait objects need a dyn before them, bare trait objects are deprecated.
So Box<std::error::Error> should become Box<dyn std::error::Error>

gardnervickers · 2019-06-11T18:23:02Z

clients/cloud/rust/src/producer.rs

+fn log_produce_result(
+    topic: &str,
+    result: Result<(i32, i64), (KafkaError, OwnedMessage)>,
+) -> Result<(), ()> {


If there is nothing to return for the result, do we even need to return anything?

I need to if I want to use it in an and_then like I do

gardnervickers · 2019-06-11T18:35:04Z

clients/cloud/rust/src/producer.rs

+        msg.wait()
+            .map_err(|err| eprintln!("error producing message: {}", err))
+            .and_then(|result| log_produce_result(&topic, result))
+            .ok();


Is this .ok() necessary?

yes because i'm using and_then which returns a Result which means that there could be an error. We know there won't be because the result we're returning doesn't throw any error but we still want to make the compiler happy

I'm not sure I follow, essentially we're going from Result<Result<(i32, i64), (KafkaError, OwnedMessage)>> here to Option<()> but not consuming the option. That should be equivalent to Result<Result<(i32, i64), (KafkaError, OwnedMessage)>> -> () (since we don't use the option. Calling expect() here would be fine since it's illegal for log_produce_result to return any error.

i'm using Result::and_then not Option::and_then so a Result will be returned, not an Option

Hmm, the .ok() here will consume the result and return an option.

yea basically I'm saying, "even if the and_then has an error in the result (which it won't), it's okay anyways". That way, the compile won't complain about an unchecked result.

gardnervickers · 2019-06-11T18:38:20Z

clients/cloud/rust/src/consumer.rs

+        msg.topic(),
+        msg.partition(),
+        msg.offset(),
+        deserialize(msg.key())?,


This could just be

msg.key() .map(|s| std::str::from_utf8(s)?) .unwrap_or("");

yea but then I have to have that same code for the value which is why I abstracted it out into a function.

ybyzek reviewed Jun 5, 2019

View reviewed changes

ybyzek requested review from ybyzek and a team June 5, 2019 11:04

ybyzek changed the title ~~[WIP] Adding Rust Client Example~~ [WIP] Confluent Cloud client example: Rust Jun 5, 2019

ybyzek reviewed Jun 5, 2019

View reviewed changes

frankgreco changed the title ~~[WIP] Confluent Cloud client example: Rust~~ Confluent Cloud client example: Rust Jun 5, 2019

ybyzek reviewed Jun 5, 2019

View reviewed changes

frankgreco force-pushed the rust branch 2 times, most recently from 08f9040 to 7fee79a Compare June 5, 2019 15:58

edenhill suggested changes Jun 5, 2019

View reviewed changes

frankgreco force-pushed the rust branch 3 times, most recently from 39372a5 to 3822821 Compare June 5, 2019 21:32

gardnervickers reviewed Jun 5, 2019

View reviewed changes

frankgreco force-pushed the rust branch 2 times, most recently from 1959e9b to 8e79558 Compare June 6, 2019 00:24

ybyzek approved these changes Jun 6, 2019

View reviewed changes

frankgreco force-pushed the rust branch 5 times, most recently from c5aa187 to 99cb68d Compare June 6, 2019 15:49

edenhill approved these changes Jun 7, 2019

View reviewed changes

gardnervickers reviewed Jun 11, 2019

View reviewed changes

adding rust client example

c6db968

frankgreco force-pushed the rust branch from 99cb68d to c6db968 Compare June 11, 2019 19:31

gardnervickers approved these changes Jun 11, 2019

View reviewed changes

ybyzek merged commit f731310 into confluentinc:5.2.1-post Jun 11, 2019

frankgreco deleted the rust branch June 11, 2019 23:18

		@@ -0,0 +1,61 @@
		# Overview

		Produce messages to produce and consume messages from [Confluent Cloud](https://www.confluent.io/confluent-cloud/) using the [rust-rdkafka client for Apache Kafka](https://github.com/fede1024/rust-rdkafka).

Confluent Cloud client example: Rust #170

Confluent Cloud client example: Rust #170

Conversation

frankgreco commented Jun 5, 2019 • edited Loading

ybyzek commented Jun 5, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

frankgreco Jun 5, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

frankgreco commented Jun 6, 2019 • edited Loading

ybyzek left a comment • edited Loading

Choose a reason for hiding this comment

edenhill left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gardnervickers left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

frankgreco commented Jun 5, 2019 •

edited

Loading

frankgreco Jun 5, 2019 •

edited

Loading

frankgreco commented Jun 6, 2019 •

edited

Loading

ybyzek left a comment •

edited

Loading