dekaf: add missing timeout to fetch_partitions() in task manager loop

[jshearer]

Summary

The Dekaf task manager is responsible for fetching and handing out fresh information about tasks that Dekaf is reading from. In order to do this, it needs to call out to various places: /authorize/dekaf, /authorize/task, and Gazette for journal info. That journal gRPC listing was the only network call in the task manager loop not wrapped in a timeout. If the RPC hangs for whatever reason, the task manager loop blocks permanently, the cached access token expires, and all sessions receive "Access token has expired and the task manager has been unable to refresh it." indefinitely until the Dekaf process is restarted.

This wraps fetch_partitions() in tokio::time::timeout(timeout, ...), matching the pattern used by get_or_refresh_dekaf_auth() and get_or_refresh_journal_client().

Incident context

Recently, we had an incident where a Gazette broker went down for reasons we are still investigating. After the dataplane recovered, 3 of 4 dekaf instances remained permanently stuck. Restarting all dekaf instances immediately resolved the issue.

The stuck instances had zero "Successful task manager run" log entries after the outage began, while the remaining healthy instance continued logging successful runs every 30 seconds. Additionally, sessions were unable to fetch tokens from the task manager due to Access token has expired and the task manager has been unable to refresh it.. This is consistent with the task manager loop being blocked on a hung journal_client.list() call inside fetch_partitions().

[jgraettinger]

I'm not sure how I feel about this. It does what it says, but I'm pretty uncomfortable with continuing to layer patches on top of task_manager.rs. I really think we need to tackle a refactor into the tokens crate (as started here)