[Kafka source] Use consumer.recv() directly rather than consumer.stream()

[guilload]

Description

Simplify tokio select loop using consumer.recv() directly rather than consumer.stream()

How was this PR tested?

Ran Kafka source on an empty topic.

[guilload]

Temporarily adding those to understand what happens when adding/removing consumers/partitions.

[kstaken]

Behind the scenes there's actually only one event in the kafka protocol, the Rust lib then adds these additional callbacks at convenient points in its handler.

[guilload]

I see. Good to know. Thanks for the explanation. One thing that I observed is that sometimes the same partitions are first revoked and then added again. Makes me question whether we should handle Revoke events.

[kstaken]

I was wondering the same but I'm pretty sure now that simple rebalances like this won't need it.

There are more advanced rebalancing strategies that try to avoid stopping the entire consumer group while partitions move but those are only really needed when you're trying to minimize latency and I don't think the Rust lib has good support for those yet. This use case also won't really need that since there's inherent latency in the ingest process no matter what.

[kstaken]

This looks good.