feat(preprod): Add confluent producer handling behind flag

[NicoHinderling]

Roll out switching traffic for preprod from the regular Producer class to the ConfluentProducer class in order to get extra producer metrics

[NicoHinderling]

• feat(preprod): Add confluent producer handling behind flag #106624 👈 (View in Graphite)
• chore(preprod): Add flagpole flag for confluent producer use #106620
• master

This stack of pull requests is managed by Graphite. Learn more about stacking.

[fpacifici]

Roll out switching traffic for preprod from the regular Producer class to the ConfluentProducer class in order to get extra producer metrics

Which additional metrics do you need ?

[NicoHinderling]

Roll out switching traffic for preprod from the regular Producer class to the ConfluentProducer class in order to get extra producer metrics

Which additional metrics do you need ?

launchpad.consumer.arroyo.producer.produce_status is the main one

[cursor]

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

^{Bugbot Autofix is OFF. To automatically fix reported issues with Cloud Agents, enable Autofix in the Cursor dashboard.}

[cursor]

Confluent producer lacks shutdown flush, risking message loss

High Severity

The new _preprod_confluent_producer global variable lacks shutdown handling. Unlike SingletonProducer, which registers an atexit handler to flush pending messages and close the producer on process termination, the Confluent producer path only calls poll(0) (non-blocking) after each produce. Messages still in the producer's internal buffer when the process exits will be silently lost. This is especially concerning since the comment indicates this is for retryable tasks, but message loss would prevent retries.

 

[cursor]

Confluent producer missing excluded config keys

Medium Severity

The original _get_preprod_producer explicitly excludes compression.type and message.max.bytes from the Kafka configuration, but _get_or_create_confluent_producer does not. The default cluster config sets compression.type: lz4 and message.max.bytes: 50000000. This means the new Confluent producer path will use different Kafka settings than the original producer, which could cause compatibility issues or unexpected behavior. This exclusion pattern is used by multiple other producers in the codebase, suggesting these keys cause problems for certain topics.

 

[getsantry]

This pull request has gone three weeks without activity. In another week, I will close it.

But! If you comment or otherwise update it, I will reset the clock, and if you add the label WIP, I will leave it alone unless WIP is removed ... forever!

---

"A weed is but an unloved flower." ― Ella Wheeler Wilcox 🥀