Event implementation

[mfontanini]

This is needed to implement the admin API (#119).

@accelerated I'm having troubles using the consumer's queue (using Consumer::get_consumer_queue) as in, I can get it out and get events out of it, but then the test gets stuck and never finishes. It looks like it's getting stuck right here. Did you have similar issues when testing this? I'm using the latest rdkafka release.

[accelerated]

@mfontanini, sorry no, I haven't seen this issue. Can you post the test case that gets stuck? I think it might be better to ask rdkafka author, he might shed more light on this. My only guess could be that it's a similar issue to the one I found where deleting the queue handle actually deletes all internal instances, instead of just decrementing the refcount. So in you case, you are deleting the queue handle and the closing the consumer which expects that queue to still be active and the events are not getting delivered anymore. As a simple test, try not deleting the queue handle and see.

[accelerated]

as a side note, you could check this into a feature branch and I can check it out and run the failing test.

[mfontanini]

This is on a separate branch so you can have a look at it. The failing test looked like this:

TEST_CASE("Event consumption", "[consumer]") {
    // Create a consumer and subscribe to the topic
    Consumer consumer(make_consumer_config());
    consumer.subscribe({ KAFKA_TOPICS[0] });

    vector<rd_kafka_event_type_t> types = {
        RD_KAFKA_EVENT_NONE
    };
    // Get consumer queue
    Queue queue = consumer.get_consumer_queue();
    for (const auto type : types) {
        const Event event = queue.next_event();
        CHECK(event.get_type() == type);
    }
}

This would just stuck forever in the consumer's destructor. I'll probably create a ticket on rdkafka, I don't think this is doing anything weird. The only thing could be the fact that it's deleting the queue but that's what the way the documentation says you should do it.

[accelerated]

Ok i saw your other test which was using a main queue so i was wondering which one it was causing the issue. I'll give it a shot later and see.

[mfontanini]

Sure, thanks. If I have some spare time at night I'll give it a shot. I feel like using rdkafka queues is very fragile. You already found that issue which would kill the queue not too long ago...

[mfontanini]

Actually, it looks like even the current test hang when running it in travis (only on one of the 2 compilers): https://travis-ci.org/mfontanini/cppkafka/jobs/444496397

[accelerated]

Ran in a loop and not a single failure with kafka version 0.11.0.2 (I added the [event] tag to single out the test).

COUNTER=0;
while [ $COUNTER -lt 500 ]; do ./cppkafka_test [consumer][event]
let COUNTER=COUNTER+1 
done

...

Running test "Event consumption" @ /bb/mbigc/mbig5696/git/cppkafka_external/tests/consumer_test.cpp:245
Done. 1/1 assertions succeeded in 1105ms
===============================================================================
All tests passed (1 assertion in 1 test case)

Running test "Event consumption" @ /bb/mbigc/mbig5696/git/cppkafka_external/tests/consumer_test.cpp:245
Done. 1/1 assertions succeeded in 1105ms
===============================================================================
All tests passed (1 assertion in 1 test case)

Running test "Event consumption" @ /bb/mbigc/mbig5696/git/cppkafka_external/tests/consumer_test.cpp:245
Done. 1/1 assertions succeeded in 1104ms
===============================================================================
All tests passed (1 assertion in 1 test case)

Running test "Event consumption" @ /bb/mbigc/mbig5696/git/cppkafka_external/tests/consumer_test.cpp:245
Done. 1/1 assertions succeeded in 1105ms
===============================================================================
All tests passed (1 assertion in 1 test case)

Running test "Event consumption" @ /bb/mbigc/mbig5696/git/cppkafka_external/tests/consumer_test.cpp:245
Done. 1/1 assertions succeeded in 1104ms
===============================================================================
All tests passed (1 assertion in 1 test case)

[mfontanini]

I think something changed on more recent kafka versions, as most tests seemed to fail once I upgraded to 2.0. Is this using the main or consumer queue? This didn't fail locally for me when using the main one.

[accelerated]

Both main and consumer queues. I ran the test once for each. I think you can safely open a bug with rdkakfa.

[accelerated]

Will you merge these changes anyway ?

[mfontanini]

I'm inclined towards merging them because this actually works, it's just that the destruction is broken, but I don't think it's cppkafka's fault. I'll probably first create a ticket on rdkafka once I can test it out a bit more and then see what to do about it.

[mfontanini]

I created confluentinc/librdkafka#2077 to keep track of the consumer getting stuck issue. For now I disabled the test, as I don't want the Travis builds to hang forever.

[mfontanini]

I'm merging this even though it seems like the round robin consumer tests are failing on rdkafka 0.9.4. I added that version of rdkafka to make sure I wasn't breaking anything when adding these changes (also by doing so I realized the library wasn't building when using 0.9.4).

I'll create an issue to tackle the failing tests. I somehow can't reproduce this locally so I'm not sure what's going on.