Using messages beyond lifetime of consumer hangs thread

[thijsc]

If you try to collect messages and use them after dropping the consumer the thread hangs indefinitely.

This test (in produce_consume_base_test.rs) reproduces the issue:

// All produced messages should be consumed.
#[test]
fn test_produce_consume_messages_beyond_consumer_lifetime() {
    let _r = env_logger::init();

    let topic_name = rand_test_topic();
    let _message_map = produce_messages(&topic_name, 100, &value_fn, &key_fn, None, None);
    let mut messages = Vec::new();

    // Drop consumer before checking vector contents
    {
        let mut consumer = create_stream_consumer(&rand_test_group(), None);
        consumer.subscribe(&vec![topic_name.as_str()]).unwrap();

        let _consumer_future = consumer.start()
            .take(100)
            .for_each(|message| {
                match message {
                    Ok(m) => messages.push(m),
                    Err(e) => panic!("Error receiving message: {:?}", e)
                };
                Ok(())
            })
            .wait();
    }

    assert_eq!(100, messages.len());
}

[fede1024]

Thanks for reporting the issue! This is probably because the correct termination sequence for librdkafka requires to destroy the messages before the consumer or the client are destroyed. This ordering is not currently enforced in rust rdkafka.

The right approach here is probably to have the lifetime of messages bound to the one of the consumer. I hope it won't make the API much more complicated.

[fede1024]

Each Message now hold a (phantom) reference to the consumer who created it. Your example won't compile anymore.