Add a few metrics on main thread contention points

[fpacifici]

Summary

After shipping the errors pipeline I saw a huge throughput problem on semantically partitioned topics with large messages. Basically this cannot keep up even with 64 replicas.
The batch time seems to fill in batches with 3k messages every second, which should be more than enough.
The parallel step is not saturated.

So the likely option is that something is really slow in the code between steps in arroyo which run on the main thread.
The sensitive areas are constituted by the calls to submit and poll.

Changes:

• Add gauge metrics and debug logs for BatchStep outbound drain (next_step.submit), including duration, outbound queue depth, and submit outcome.
• Add gauge metrics and debug logs for PythonAdapter Python delegate submit, poll/handle paths, and downstream next_strategy.submit.
• Reuse cached metric labels on both steps to avoid per-call allocations.

Made with Cursor