feat(kafka): Add Arroyo consumer backend

[lvthanh03]

Adds the TASKBROKER_USE_ARROYO_CONSUMER config. When it is set, the broker consumes with Arroyo's AsyncKafkaConsumer instead of rdkafka's StreamConsumer.

Since it is a static config, switching consumer backend requires a restart, but I think the tradeoff of extra simplicity on the consumer is worth it as we avoid having to tear down the running consumer mid-process, and avoid introducing the runtime config lifecycle into the consumer code.

Implemented:

• ConsumerBackend wrapping both clients. It creates them from the same client config and hands out partition queues.
• PartitionQueueBackend wrapping rdkafka's StreamPartitionQueue and Arroyo's AsyncPartitionQueue behind MessageQueue, so the map actors stream messages from either.
• Arroyo's on_assign and on_revoke send the same events on the same channel as rdkafka's pre_rebalance, and block until handle_events completes the rendezvous. handle_events and processing_strategy! are shared.

Some more context:

• Arroyo polls on its own thread and commits when it is dropped, so it doesn't use poll_consumer_client. On revoke and shutdown, its aborted actors are joined, so none stores offsets after the revoke or still holds the consumer when it closes.
• start_consumer drops the consumer (for both rdkafka and Arroyo consumers) in spawn_blocking after handle_events returns, so the blocking close and final commit stay off the async workers.
  • In the case of rdkafka, the consumer blocks while it closes too, so this only moves that close off the async workers when start_consumer holds the last reference.

CI was also updated to run on both consumer backends.

Refs STREAM-1859

[linear-code]

STREAM-1859

[cursor]

Cursor Bugbot has reviewed your changes and found 3 potential issues.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 2e4ed3e. Configure here.}

[evanh]

Mostly looks good, couple style things.

[evanh]

I think I would rather this be a Trait, and then have the logic for the two backends split up, instead of having if statements throughout the functions. Is that possible?

[lvthanh03]

It's possible but since we eventually want to be on the Arroyo consumer, I don't think we need an abstraction around a backend we're going to remove. The backend is also picked at runtime, so a trait would just move the branch to main.rs (start_consumer::<ArroyoConsumer> vs start_consumer::<RdkafkaConsumer>)

[untitaker]

is upkeep even related to kafka consumer backends?