Added pause/resume for producers

[accelerated]

Added similar helper functions to consumer::pause and consumer::resume.

[accelerated]

Not sure if it's worth caching the Topic object/partition list inside the producer so it can be reused when resume is called?

[mfontanini]

I would just use a forward declaration here, this pulls in a bunch of crap that I tried avoided pulling in other headers (e.g. kafka_handle_base.h)

[accelerated]

ok will do

[mfontanini]

Woah I didn't know pausing works for producers... Do you have a use case and do you know what happens when you produce into a paused topic/partition? Does it just get buffered for a while?

I feel like this should have an overload that takes TopicPartitions. I'm not a super fan of get_metadata but maybe I'm biased because I've faced at least 2 bugs in rdkafka related to this. Anyway, I imagine someone may want to not produce into some particular topic/partition although I'm still not sure of what the use case would be for it.

[accelerated]

Well apparently the rdkafka documentation for pause/resume does not involve a handle at all and it specifically refers to both production and consumption. For producers you pause when you get throttled. I'm currently implementing some throttling logic on top of cppkafka so i want to be able to pause on both ends. For the producer side, i assume that if you pause, the internal rdkafka queue won't get flushed at all, thus leaving the broker some time to breathe. During this time, all sync/async_produce() get routed to add_message instead which buffers locally in the BufferedProducer. When the throttle comes off, I resume and flush the buffer.
No need to add an overload for TopicPartitions, you already have that function in KafkaHandleBase which takes a list. However I wanted to add an overload in the consumer to take a topic name. One use case is that i want to start some consumers in "paused" state and only start consuming when i'm ready. This is even before partitions get assigned so I would need a way to pause a topic rather than an "assigned partition list".

[accelerated]

I also assume that when pausing the consumer while continuing to poll it, only heartbeats go out and no more batch message requests.

[accelerated]

Actually the pause(topic), resume(topic) should go in the KafkaHandleBase as the implementation is generic to both. I'll push a change asap.

[accelerated]

BTW posted a question for clarification in librdkafka

[mfontanini]

Interesting. The author's answer regarding pausing unassigned partitions makes me wonder if this is a valid change. Like he says it may work but they don't guarantee it does. Let's wait for him to answer your last question.

[accelerated]

Tested with the PR pause/resume code as indicated by librdkafka owner and it works fine. Essentially getting the metadata of the partitions makes it known locally and then pausing it works as indicated. I could not use it inside the assignment callback because in cppkafka you call assign(handle, partitions) after the user registered assignment callback is raised. Therefore, unless i call assign twice it cannot be done.

[accelerated]

Cheers!

[mfontanini]

Thanks for this one!