fix(run_task_threads): Add metric to capture backpressure

[nikhars]

In order to debug backpressure being raised from run task in threads, add a new metric which captures the information. This would help triage whether the kafka doomsday alert is going off because of backpressure or not.

[lynnagara]

Not sure I follow why this is needed for debugging that specific alert. Why do you want the count of message rejected (as opposed to time spent)? arroyo.processor.paused.time gives you the time the processor spends in this state, which is what we generally use instead.

[nikhars]

Not sure I follow why this is needed for debugging that specific alert. Why do you want the count of message rejected (as opposed to time spent)? arroyo.processor.paused.time gives you the time the processor spends in this state, which is what we generally use instead.

I was emulating what the run task with multiprocessing does over here. But you are right, I can use the metric you provided to see if the consumer gets paused or not. Will close out this PR.