Fix AwaitMessageTrigger crash on tombstone messages when apply_function is unset

[TimurRakhmatullin86]

When AwaitMessageTrigger is used without an apply_function (supported since #55437), the trigger calls message.value().decode("utf-8") on every polled message. For a tombstone message — value() is None, a routine occurrence on log-compacted topics where a null value is the deletion marker — this raises AttributeError: 'NoneType' object has no attribute 'decode', crashing the trigger and failing the deferred task.

This PR treats a tombstone the same way as any other non-matching message: no event is emitted, the offset is committed (when commit_offset=True, matching the existing non-matching path so the tombstone is not re-read forever after a restart), and the trigger keeps polling.

Why skip rather than emit: the trigger's contract is already truthiness-based (if event:), so a None payload cannot be yielded on this path today, and a TriggerEvent(None) would carry no usable payload downstream. Users who need to react to tombstones can already do so via apply_function, which receives the raw message (including value() is None) and can map it to any truthy payload. Surfacing tombstones from the no-apply_function convenience path would be a separate opt-in feature; crashing on them is a bug.

Tests:

• test_trigger_run_tombstone_message_keeps_polling — regression test: poll() returns a null-value message; asserts the trigger neither raises nor yields and commits the tombstone offset. Fails on main with the exact AttributeError above.
• test_trigger_run_without_apply_function_yields_message_value — asserts the happy path still yields the decoded payload.
• MockedMessage now has a value() method: the apply_function=None case of test_trigger_run_good previously passed for the wrong reason — the mock lacked value(), so the task finished via this same AttributeError, satisfying task.done() is True.

No behavior change for non-null messages or when apply_function is set. Empty (b"") payloads were already treated as non-events by the if event: check — unchanged.

[potiuk]

Good fix for a genuinely routine case — a null value on a log-compacted topic is the deletion marker, not an anomaly, so crashing the trigger on it is a real problem rather than an edge case.

The part I like is that it routes tombstones into the existing non-matching branch rather than special-casing them. That branch already commits the offset when commit_offset is enabled, which matters more than it might look: without it a tombstone sitting at the head of a compacted topic would be re-read on every restart and the trigger would never progress past it. Worth having pinned in the test, which you did:

assert commit_mock.mock_calls == [call(message=message, asynchronous=False)]

alongside assert task.done() is False. The latter is what fails on main, where the AttributeError completes the task.

One thing worth being aware of rather than changing here: a non-null but empty payload (b"") decodes to "" and is also skipped, since the contract is truthiness-based. That is pre-existing behaviour and this PR does not alter it, but it means "no event emitted" covers slightly more than just tombstones.

Rebased onto main before merging, as the branch had drifted well behind.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting