Investigate use of split_partition_queue to simplify StreamProcessor

[untitaker]

getsentry/taskbroker is using split_partition_queue, and that allows them to poll the consumer separately from pulling messages out of it.

Right now we pause the consumer on backpressure to not receive messages. If we used split_partition_queue, that would be obsolete and handling backpressure in arroyo would become simpler.

It seems however that this is not an option for python, as confluent-kafka doesn't have split_partition_queue. Might be a Rust-only thing.