This gets started on #50, #51. I've built what I believe to be the appropriate wrapper around the connections now I'm just trying to figure out the best way to expose it. Would love to hear your feedback @spicavigo.

It was mostly inspired by: https://github.com/hyperium/hyper/blob/master/src/net.rs#L655-L664

cc @erickt

Resolves #128

• [x] Make KafkaClient.fetch_group_(topic_)?offsets return a Result<HashMap<String, Vec<PartitionOffset>>>
• [x] Make PartitionOffset.offset an i64 rather than Result<i64>
• [x] Change KafkaClient::produce_messages to return a new struct, ProduceConfirm, rather than TopicPartitionOffset, and remove TopicPartitionOffset

Unfortunately, I have not had time the last few years to continue maintenance of this project, nor has @xitep or @spicavigo. Please post here if you are interested in becoming a maintainer.

I'm trying to use this lib to calculate the offset lag for a particular group ID. It seems able to fetch topic metadata though KafkaClient::fetch_offsets just fine, but when specifying a group with KafkaClient::fetch_group_topic_offsets, it returns an UnknownTopicOrPartition error in the resulting offset entry.

fn main() {
    ...
    let kafka_config = config.kafka_config;

    let mut client = KafkaClient::new(kafka_config.broker_list.clone());
    client.load_metadata(&kafka_config.topics).unwrap();

    let offsets = client.fetch_offsets(&kafka_config.topics, FetchOffset::Latest)
        .unwrap();
    println!("{:#?}", offsets);

    for topic in kafka_config.topics {
        let group_offsets = client.fetch_group_topic_offsets(&kafka_config.group, &topic)
            .unwrap();
        println!("{:#?}", group_offsets);
    }
}

this prints out

{
    "foo": [
        PartitionOffset {
            offset: Ok(
                299905
            ),
            partition: 3
        },
        PartitionOffset {
            offset: Ok(
                299905
            ),
            partition: 6
        },
    ...
}
[
    TopicPartitionOffset {
        offset: Err(
            Kafka(
                UnknownTopicOrPartition
            )
        ),
        topic: "foo",
        partition: 3
    },
    TopicPartitionOffset {
        offset: Err(
            Kafka(
                UnknownTopicOrPartition
            )
        ),
        topic: "foo",
        partition: 6
    },
    ...
]

These are active Kafka (v0.9) topics that are known to be working with the official Java library, and have committed offsets. It's also worth mentioning that when I create a Consumer and poll messages on the same group ID, it does not find a committed offset for the group, even though it does have one--it falls back to the offset specified by the fallback option.

I'm a beginner of kafka and rust, to use rust client as a consumer, I add the github example in src/main.rs

extern crate kafka;
use kafka::client::KafkaClient;
fn main() {
    let mut client = KafkaClient::new(&vec!("localhost:9092".to_string()));
    client.load_metadata_all();
    // OR
    // client.load_metadata(&vec!("my-topic".to_string())); // Loads metadata for vector of topics
 }

and add kafka = "*" in the cargo.toml dependencies

then I run $ cargo build I have the following error, is there anyone can help me to solve this?

   Compiling kafka v0.1.6 (file:///Users/admin/myrustdemo/kafka-rust-master)
src/main.rs:2:1: 2:20 error: multiple rlib candidates for `kafka` found
src/main.rs:2 extern crate kafka;
              ^~~~~~~~~~~~~~~~~~~
src/main.rs:2:1: 2:20 note: candidate #1: /Users/admin/myrustdemo/kafka-rust-master/target/debug/libkafka.rlib
src/main.rs:2 extern crate kafka;
              ^~~~~~~~~~~~~~~~~~~
src/main.rs:2:1: 2:20 note: candidate #2: /Users/admin/myrustdemo/kafka-rust-master/target/debug/deps/libkafka-73c7ae9da4235a7d.rlib
src/main.rs:2 extern crate kafka;
              ^~~~~~~~~~~~~~~~~~~
src/main.rs:2:1: 2:20 error: can't find crate for `kafka`
src/main.rs:2 extern crate kafka;
              ^~~~~~~~~~~~~~~~~~~
error: aborting due to 2 previous errors
Could not compile `kafka`.

to solve #149

there are some limitation:

• it's hard to impl Clone for Error
• native-tls only support to load CA in DER format and Cert/PKey in PKCS12 format, sfackler/rust-native-tls#27
• native-tls doesn't support disable verification, sfackler/rust-native-tls#3
• native-tls doesn't support set_cipher_list, sfackler/rust-native-tls#4

The "key" - if available - of a message is typically used to derive a consistent partition which a message should be delivered to. Usually one computes a hash code of the key and maps that consistently on one of the partitions of the topic.

Clients of the Producer API can achieve this already by supplying a custom Partitioner to the Producer. However, we shall make this the standard behaviour of the DefaultPartitioner.

Unfortunately, topic metadata held by the Producer (which is supplied by KafkaClient) deliver only "available" partitions, which makes it impossible to implement the "consistency" aspect of the requirement. Therefore, this will require an extension to KafkaClient as well.

Regarding the hash algorithm we don't need a cryptographic hash function. However, we need to use one which doesn't different depending the platform - this rules out farmhash for example. (it seems the java based producer uses murmur2-32bit, but i don't see the need to use the same since various kafka clients seem not to stick a "default" hash function, e.g. the java based producer uses a different hash function than the scala based one.)

I'm trying to use a Rust consumer to read from multiple topics. This is the code I have now:

extern crate kafka;
use kafka::client::KafkaClient;
use kafka::consumer::Consumer;
use kafka::utils;
fn main(){
    let mut client = KafkaClient::new(vec!("localhost:9092".to_owned()));
    let res = client.load_metadata_all();
    let topics = client.topic_partitions.keys().cloned().collect(); 
    let offsets = client.fetch_offsets(topics, -1);
    for topic in &topics {
    let mut con = Consumer::new(client, "test-consumer-group".to_owned(), "topic".to_owned()).partition(0);
    let mut messagedata = 0;
    for msg in con {
        println!("{}", str::from_utf8(&msg.message).unwrap().to_string());
    }
  }
}

below is the error:

    src/main.rs:201:19: 201:25 error: use of moved value: `topics` [E0382]
src/main.rs:201     for topic in &topics {
                                  ^~~~~~
    note: in expansion of for loop expansion
    src/main.rs:201:5: 210:6 note: expansion site
    src/main.rs:167:40: 167:46 note: `topics` moved here because it has type `collections::vec::Vec<collections::string::String>`, which is non-copyable
    src/main.rs:167     let offsets = client.fetch_offsets(topics, -1);
                                                       ^~~~~~
    src/main.rs:203:37: 203:43 error: use of moved value: `client` [E0382]
    src/main.rs:203     let mut con = Consumer::new(client, "test-consumer-group".to_owned(), "topicname".to_owned()).partition(0);
                                                    ^~~~~~
    note: in expansion of for loop expansion
    src/main.rs:201:5: 210:6 note: expansion site
    note: `client` was previously moved here because it has type     `kafka::client::KafkaClient`, which is non-copyable
    error: aborting due to 2 previous errors

To better explain my question, here is my partial workable code for just one topic:

let mut con = Consumer::new(client, "test-consumer-group".to_owned(), "testtopic".to_owned()).partition(0);

for msg in con {
    println!("{}", str::from_utf8(&msg.message).unwrap().to_string());
}

And I tested the fetch_message function, it works for multiple topics, but the result I have (msgs) is Topicmessage, I don't know how to get message from Topicmessage.

let msgs = client.fetch_messages_multi(vec!(utils::TopicPartitionOffset{
                                            topic: "topic1".to_string(),
                                            partition: 0,
                                            offset: 0 //from the begining
                                            },
                                        utils::TopicPartitionOffset{
                                            topic: "topic2".to_string(),
                                            partition: 0,
                                            offset: 0
                                        },
                                        utils::TopicPartitionOffset{
                                            topic: "topic3".to_string(),
                                            partition: 0,
                                            offset: 0
                                        }));
for msg in msgs{
    println!("{}", msg);
}

Currently, the utils::{TopicPartitionOffset, PartitionOffset} structs' offset fields are Results. In my opinion, it's kind of strange to have a struct with a Result in the middle of it for anything but maybe custom Error types. It can make error handling on the user-side really inconvenient.

Additionally, it's not entirely clear to me why TopicPartitionOffset is needed at all—it seems that it exists only to flatten responses, which leads to lots of inefficient cloning of data that it already had. I think in an instance like the linked one, it would be more appropriate to return a map which maps topics to PartitionOffsets; that way, only a move of one string is necessary, rather than a clone for every single offset response.

What do you think? I tried to start by changing the Results to just their underlying type, but I can see it's a non-trivial refactor, so I thought I'd get your input first.

Hey there,

I'm working on a Heroku application that will be utilizing Kafka and am trying to build this library. As part of this I have a script which builds the libsnappy.a file by make installing Snappy into $CACHE_DIR/snappy-build/lib, here is the script:

#!/usr/bin/env bash
# bin/compile <build-dir> <cache-dir> <env-dir>

set -e
set -o pipefail

# Build related variables.
BUILD_DIR=$1
CACHE_DIR=$2
ENV_DIR=$3

if [[ ! -d "$CACHE_DIR/snappy" ]]; then
    mkdir -p $CACHE_DIR
    cd $CACHE_DIR
    git clone https://github.com/google/snappy
    cd snappy
    ./autogen.sh
    ./configure --prefix="$(pwd)/../snappy-build"
    make
    make install
fi

export LD_LIBRARY_PATH="$CACHE_DIR/snappy-build/lib"
export LD_RUN_PATH="$CACHE_DIR/snappy-build/lib"

However whenever I try to push my application the kafka-rust library tries to link it incorrectly. See:

src/main.rs:5:5: 5:31 warning: unused import, #[warn(unused_imports)] on by default
src/main.rs:5 use kafka::client::KafkaClient;
                  ^~~~~~~~~~~~~~~~~~~~~~~~~~
error: linking with `cc` failed: exit code: 1
note: "cc" "-Wl,--as-needed" "-m64" "-L" "/root/.multirust/toolchains/nightly/lib/rustlib/x86_64-unknown-linux-gnu/lib" "/source/target/debug/kafka_rust_test.0.o" "-o" "/source/target/debug/kafka_rust_test" "-Wl,--gc-sections" "-pie" "-nodefaultlibs" "-L" "/source/target/debug" "-L" "/source/target/debug/deps" "-L" "/source/target/debug/build/miniz-sys-d03126dbc9ee0074/out" "-L" "/source/target/debug/build/openssl-bade188098f75a08/out" "-L" "/usr/lib" "-L" "/source/target/debug/build/openssl-sys-extras-4daf08572eefce4c/out" "-L" "/root/.multirust/toolchains/nightly/lib/rustlib/x86_64-unknown-linux-gnu/lib" "-Wl,-Bstatic" "-Wl,-Bdynamic" "/source/target/debug/deps/libkafka-9ca67eb32e4e1893.rlib" "/source/target/debug/deps/libiron-4fdc411e735a87b7.rlib" "/source/target/debug/deps/libmodifier-550b2f0e350e2963.rlib" "/source/target/debug/deps/liblibc-dd3420cb049117bb.rlib" "/source/target/debug/deps/libhyper-6df9e9da10e4a36f.rlib" "/source/target/debug/deps/libunicase-4ad2965620fe21a9.rlib" "/source/target/debug/deps/libsolicit-8632432b3a4330d6.rlib" "/source/target/debug/deps/libhttparse-5c82294627258d33.rlib" "/source/target/debug/deps/libhpack-8f91a695370f3d75.rlib" "/source/target/debug/deps/libplugin-99f6a9d43520a61a.rlib" "/source/target/debug/deps/liberror-d27f5056e37dd662.rlib" "/source/target/debug/deps/liblanguage_tags-9df4f269b9ee12d4.rlib" "/source/target/debug/deps/libnum_cpus-9a6b3f359403ec12.rlib" "/source/target/debug/deps/libbyteorder-3945c3ad3e0e1d9c.rlib" "/source/target/debug/deps/libtraitobject-4ea485452a3a4a0b.rlib" "/source/target/debug/deps/libtypemap-487ec1564575dd9b.rlib" "/source/target/debug/deps/libconduit_mime_types-8c1fe30d92f8233a.rlib" "/source/target/debug/deps/libmime-f5540d134e188f4e.rlib" "/source/target/debug/deps/liblog-87d547eff707fc8e.rlib" "/source/target/debug/deps/libcookie-9ec7d33888fc3f77.rlib" "/source/target/debug/deps/liburl-b7bb31aacec501cd.rlib" "/source/target/debug/deps/libuuid-fed17b74aa7673e2.rlib" "/source/target/debug/deps/libmatches-737aa40e66529b02.rlib" "/source/target/debug/deps/libopenssl-bade188098f75a08.rlib" "/source/target/debug/deps/liblazy_static-007034d2ad8108ce.rlib" "/source/target/debug/deps/libbitflags-646076c1f4684754.rlib" "/source/target/debug/deps/libopenssl_sys_extras-4daf08572eefce4c.rlib" "/source/target/debug/deps/libopenssl_sys-5ec15f2e42328238.rlib" "/source/target/debug/deps/libflate2-1e8d03096e238ad4.rlib" "/source/target/debug/deps/libminiz_sys-d03126dbc9ee0074.rlib" "/source/target/debug/deps/libserde-a8734c231f4c36ee.rlib" "/source/target/debug/deps/libnum-02177b937f857300.rlib" "/source/target/debug/deps/librand-340832a8942cb900.rlib" "/source/target/debug/deps/librustc_serialize-3c33cb2a40992011.rlib" "/source/target/debug/deps/libtypeable-7ddee84661471c9b.rlib" "/source/target/debug/deps/libtime-22c21fe32894ddad.rlib" "/source/target/debug/deps/liblibc-adb8b8e7aaa2f93f.rlib" "/source/target/debug/deps/libunsafe_any-081a220f4ebf6660.rlib" "/source/target/debug/deps/libtraitobject-a729e3ae66a1ef57.rlib" "/root/.multirust/toolchains/nightly/lib/rustlib/x86_64-unknown-linux-gnu/lib/libstd-17a8ccbd.rlib" "/root/.multirust/toolchains/nightly/lib/rustlib/x86_64-unknown-linux-gnu/lib/libcollections-17a8ccbd.rlib" "/root/.multirust/toolchains/nightly/lib/rustlib/x86_64-unknown-linux-gnu/lib/librustc_unicode-17a8ccbd.rlib" "/root/.multirust/toolchains/nightly/lib/rustlib/x86_64-unknown-linux-gnu/lib/librand-17a8ccbd.rlib" "/root/.multirust/toolchains/nightly/lib/rustlib/x86_64-unknown-linux-gnu/lib/liballoc-17a8ccbd.rlib" "/root/.multirust/toolchains/nightly/lib/rustlib/x86_64-unknown-linux-gnu/lib/liballoc_jemalloc-17a8ccbd.rlib" "/root/.multirust/toolchains/nightly/lib/rustlib/x86_64-unknown-linux-gnu/lib/liblibc-17a8ccbd.rlib" "/root/.multirust/toolchains/nightly/lib/rustlib/x86_64-unknown-linux-gnu/lib/libcore-17a8ccbd.rlib" "-l" "snappy" "-l" "c" "-l" "m" "-l" "ssl" "-l" "crypto" "-l" "dl" "-l" "pthread" "-l" "gcc_s" "-l" "pthread" "-l" "c" "-l" "m" "-l" "rt" "-fuse-ld=gold" "-l" "compiler-rt"
note: /usr/sbin/ld.gold: error: cannot find -lsnappy

Could you perhaps offer some guidance? I've read through the build.rs script in this repo and have been trying to convince it unsuccessfully.

I create a topic with 1 partition, and use the following code to create some consumers:

Consumer::from_hosts(brokers)
    .with_topic("test".to_owned())
    .with_group("group-recognizer".to_owned())
    .with_fallback_offset(FetchOffset::Latest)
    .create()
    .unwrap()

then use the following code to listen the topic's message:

for ms in consumer.poll().unwrap().iter() {
    for m in ms.messages() {
        println!("{:?}", m);
    }
    consumer.consume_messageset(ms);
}
consumer.commit_consumed().unwrap();

Also, I notice that no output like the following when those consumers start:

[2017-06-21 00:06:55,868] INFO [GroupCoordinator 0]: Assignment received from leader for group group-recognizer for generation 4 (kafka.coordinator.GroupCoordinator)

which will be generated when using kafka-console-consumer.sh with the same group.id in properties file.

Edit: When add .with_offset_storage(GroupOffsetStorage::Kafka) to Consumer builder, the program panic on when receiving the second message:

thread 'main' panicked at 'called `Result::unwrap()` on an `Err` value: Kafka(UnknownMemberId)', src/libcore/result.rs:860
note: Run with `RUST_BACKTRACE=1` for a backtrace.

I am using the kafka library and have multiple replicas of my consumers to ensure HA. However, I have noticed that even that I specify the consumer group and the client id, that the replicas are getting the same messages from kafka.

After logging the consumer partitions, I noticed that they are the same.

Therefore, I just want to confirm if it is possible to have multiple consumers for the same topic, consuming messages from different partitions without I manually specifying which partition should be consumed. My goal is to avoid the same messages to be processed multiple times by different consumers.

After reading through the protocol code, I thought it needed a bit of a rework - there is room for improvement (remove unsafe code, more idiomatic rust, and better performance)

I played around a bunch with alternative designs for better performance (notably for decoding) but it may be too complicated to adopt one of those approaches right now. Essentially, there should be custom read that allows to retrieve the bytes that are being read, which would minimize allocations and copies (see bincode).

For now I will try to implement something that works and that fully implements the protocol using 0 unsafe code and idiomatic Rust, I will look into the custom Read later on

Integration Tests

• Added Kafka 3.1.0
• Updated readme to reflect versions tested
• Simplified Dockerfile
• Added set -e to potentially catch other errors

Dependencies

• Switched logs to use 'tracing' crate instead
• Removed 'log' and 'env_logger' dependencies
• Added 'tracing-subscriber' dependency to replace 'env_logger'

Core

• Made offset storage optional
• Error during commit if offset storage or group id is empty

Examples (esp. console-consumer)

• Added --follow option
• Changed '--no-commit' to '--commit'
• Default to 'GroupOffsetStorage=None'
• etc.