Add PartitionQueue forwarding

[matyama]

Hi!

This PR is proposing a small extension to the PartitionQueue, or rather it just exposes the unsafe rd_kafka_queue_forward binding in a safe manner on the PartitionQueue.

Currently, the BaseConsumer can split its main queue into partition queues, but it comes quite handy to have an option to "join" these splitted queues back. For instance, one can have a pool of worker threads, each assigned with a subset of partitions of a topic, and then it comes quite handy to have the worker just forward all the queues into one and poll it (this is previewed in the tests).

To prevent misuse, the added forward function first checks that both involved consumers (of the forwarded and destination queues) point to the same underlying client. In case they don't, a newly introduced KafkaError::ClientMismatch is returned to inform about this problem. I've decided to add a new type because none of the existing ones seem quite fitting for this.

[matyama]

Hi, I've added a small update that introduces a PartitionQueue::reset_forwarding method. This one is complementary to the forward one and forwards a partition queue back to the main consumer queue.

Also, would there be a chance to fit this into one of the next releases (after a review etc. of course)? I'm currently depending on the fork, which is not ideal :)

[davidblewett]

Hmm, it might be surprising to have reset_forwarding send this queue back to the main consumer queue? I would have expected it to be the inverse of forward, where any previous forwarding would be removed and this queue will resume receiving messages. This is the behavior when calling rd_kafka_queue_forward with a NULL pointer for the dst queue.

[matyama]

I agree that the naming might be unfortunate and misleading. IIRC the motivation here was to provide a method to go back ("reset") to the main consumer queue as the current public API does not expose the consumer queue (i.e., it can't be passed to forward as the dst).

But it totally makes sense to have a way how to clear the forwarding as well. I think there are several ways how to approach this:

1. There could be two methods to reset/clear forwarding - reset being the current one and clear_forwarding would use the null ptr.
2. Expose a ConsumerQueue and change reset_forwarding as you propose (and generalize forward). But I'm not quite sure if there's enough value in adding such a public wrapper for the main queue.
3. Stick closer to the bindings and overload forward for all the forwarding modes. Something along these lines:

   // impl PartitionQueue<C>
   pub fn forward<'a>(&mut self, mode: impl Into<ForwardMode<'a, C>>) -> KafkaResult<()> {
       match mode.into() {
           ForwardMode::Clear => todo!("forward using null ptr"),
           ForwardMode::Consumer => todo!("forward to consumer queue"),
           ForwardMode::Queue(dst) => todo!("forward to dst"),
       }
   }
   
   // describes how/where to forward
   pub enum ForwardMode<'a, C: ConsumerContext> {
       // -> ptr::null_mut()
       Clear,
       // -> self.consumer.client().consumer_queue()
       Consumer,
       // -> this PartitionQueue
       // with an impl From<&mut 'a PartitionQueue<C>>
       Queue(&'a mut PartitionQueue<C>),
   }

   I quite like this option, but what's a bit sad here is that Clear and Consumer would still return a KafkaResult even though they should/could be infallible.
4. Remove reset_forwarding altogether and leave this feature for future PRs. My use-case was rather static (i.e., forward and forget), I just thought that some form of reset might be useful in a more dynamic setup.

[davidblewett]

I like the 3rd option the best. If you don't have a specific use case for it however, I'm wondering if a simpler API would be to just update the Drop implementation for PartitionQueue to forward the queue back to the main consumer queue?

[matyama]

I've considered implementing the reset inside Drop before, but the issue here is that it may produce unexpected side-effects when one wants to just forward a queue and then drop it - for instance:

start with a 3-way split (q0, q1, q2)
forward q0 -> q1  // now we expect q1 to receive (q0 + q1)
drop(q0)          // but here q0 returns back to main due to the implicit reset
// now we have (q0 -> main, q1, q2), which might be confusing

This could be prevented by keeping these "source" queues around, but that's not great either.

[matyama]

So I went with the 3rd option (the generalized forward) in the latest commit and added couple more tests ;)