Shash sriv/stream 1807 discard tasks from incorrect applications

[ShashSriv]

A pool can only run tasks for the applications its workers serve. Nothing checked
that, so a misrouted task never died:

• pull mode nobody claimed it and it sat inthe DB until it paged us
• push mode it was claimed, found no worker, and got re-fetched forever.

This pr allows the consumer to drop it before it's ever stored: metric, error log, and a copy
published to the pool's deadletter topic so it's recoverable. Tasks already in the
DB get the same treatment via the push thread setting Failure, which hands them to
upkeep's existing discard/deadletter path.

Push pools derive the allowed set from worker_map and are covered on deploy. Pull
pools have no worker_map, so they stay unrestricted until ops sets applications
on them — follow-up PR in ops. Nothing changes behaviour until then.

[linear-code]

STREAM-1807

[cursor]

Stale Bugbot comment from a previous run.

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 7bf2633. Configure here.}

[sentry]

Bug: Deadlettered tasks are lost if publishing to the deadletter topic fails, as Kafka offsets are committed regardless of the publish outcome.
_{Severity: HIGH}

Suggested Fix

After calling send_all, check if the number of successes equals the number of attempts. If they do not match, do not clear the deadletter_batch and return a result that prevents the consumer from committing the Kafka offsets. This will ensure that failed messages can be reprocessed.

Prompt for AI Agent

Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: src/kafka/activation_batcher.rs#L327

Potential issue: In the `flush()` method's deadletter handling, tasks are published to a
deadletter topic via `send_all`. However, the code does not verify if all publish
attempts were successful. It unconditionally clears the `deadletter_batch` on line 327
and returns `Ok(Some(...))`. This successful return value causes the upstream Kafka
consumer to commit the message offsets. In a scenario where publishing to the deadletter
topic partially fails (e.g., due to transient network issues), the failed messages are
permanently lost. They are not in the database, failed to reach the deadletter topic,
and will not be re-consumed from Kafka because their offsets have been committed.

[ShashSriv]

Offsets commit per inflight batch, not per message, so the only way to avoid committing is to fail the flush which sends the batch down the reducer error path. A brief DLQ hiccup would stall the whole topic, which is worse than dropping a task this pool can't run anyway.

The send waits out kafka_send_timeout_ms with librdkafka retrying underneath. This also matches what the demoted-namespace forward path already does.

We'll alert on consumer.unknown_application_failures instead.