[tryangul]

What

Clean up pull consumer wiring: shared resources, factory extraction, module split, and upstream arroyo alignment.

How

Wiring

• SharedResources — Kafka producers (DLQ, commit log, COGS) and CH writer created once, Arc-shared across rebalance iterations
• Factory functions — make_kafka_source, make_dlq_handler, make_ch_writer, etc. replace inline construction
• Module split — pull_consumer.rs → pull_consumer/ with entrypoint.rs (bootstrap/runtime), factories.rs (component construction), pipelines.rs (pipeline assembly)

Rebalance / rebuild improvements

• Owned Pipeline::stream(self, ...) — pipelines consume themselves, dropping the 'a lifetime
• PipelineRunner — replaces the hand-rolled rebalance loop

Upstream API changes

• KafkaProducer::new() → Result — upstream arroyo#563 changed the return type; added .expect() at all call sites

Why

• Rebalance recovery was re-creating 3-4 Kafka producers + a CH client every time — wasteful and slow
• Wiring was one 530-line function; modules make it navigable and testable
• Pull model isn't enabled anywhere yet, so this is zero-risk

Notes

• Depends on arroyo PR #560 — Cargo.toml tracks the branch; will pin to a rev once merged
• The KafkaProducer::new .expect() changes in consumer.rs, factory_v2.rs, accepted_outcomes_consumer.rs are required by the upstream API change, not specific to pull

Links

• Arroyo PR: feat(runtime): Implement pull based arroyo arroyo#560

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

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

Want reviews to match your repository better? Bugbot Learning can learn team-specific rules from PR activity. A team admin can enable Learning in the Cursor dashboard.

^{Reviewed by Cursor Bugbot for commit 4ef6dd3. Configure here.}