[DNM] Add Arroyo batch timeout logic

[trevor-e]

Goal: we are processing third-party apps and cannot predict the input will always cleanly succeed, it's fine and expected that some processing jobs will fail. We do need some guarantees that jobs will only take a certain amount of time and not clog up the system for everyone else, though.

For context, we ran into an edge case where our lief library essentially deadlocked our topic. It wasn't technically deadlocked, but it would have taken several hours to finish processing the message. Fixing this was rather painful in prod once it happened since we had to skip over that message. Instead, we want a hard processing timeout per message.

I originally looked into supporting timeouts per message batch in Arroyo itself, however this runs into an undesirable case where we don't know exactly which messages in a batch succeeded or failed and we might erroneously mark successful messages to the DLQ. This might be fine, but I'm not sure of the idempotency guarantees Arroyo makes. In our case this doesn't matter since we always set the batch size to 1, but since this is a general purpose library I could see this being confusing/unexpected behavior to others. The streaming team mentioned we could have per-message timeouts but it might be a larger undertaking/refactor to achieve that.

Instead, in the meantime we can implement timeouts in our subclass. This works by overriding the poll() method, recording the start times of batches, and then checking for batches exceeding our timeout threshold. Since we are the only team requesting this feature we can live with this for now until it's potentially supported upstream. I tried a few other methods such as running the processing logic in a thread with a join timeout, but those never trigger.

Testing this with the stuck build in question I now correctly see:

[17:22:13] ERROR Batch exceeded timeout of 720.0s (elapsed=720.70s).
DEBUG Worker process exited normally (SIGCHLD 20)
DEBUG Worker process exited normally (SIGCHLD 20)
INFO Terminated worker and sent 1 messages to DLQ

[trevor-e]

This is brittle but we'll have alerting if something breaks around this. Timed out messages are pretty rare and should have plenty of time to react in this case.

[trevor-e]

This part I wasn't entirely sure about, but I didn't think I could raise InvalidMessage from this part of the code.

[codecov]

Codecov Report

❌ Patch coverage is 6.97674% with 40 lines in your changes missing coverage. Please review.
✅ Project coverage is 80.06%. Comparing base (ebde423) to head (6d43688).
⚠️ Report is 1 commits behind head on main.
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
src/launchpad/kafka.py	6.97%	40 Missing ⚠️

Additional details and impacted files

@@ Coverage Diff @@## main #461 +/- ##
==========================================
- Coverage 80.26% 80.06% -0.21% 
==========================================
Files 159 159 Lines 13482 13524 +42 Branches 1428 1434 +6 ==========================================
+ Hits 10822 10828 +6 - Misses 2122 2158 +36 
Partials 538 538 

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[chromy]

stamping in case we want to land need to land this in order to timeouts working, as discussed DMs maybe we're not using a lot of RunTaskWithMultiprocessing as this point (we're not using batching or pools and the timeout logic has to reach into the internals) - but then there is a some book keeping if we make our own Strategy.