A consumer that stopped on a poison message, on purpose

A message that cannot be processed and must not be skipped, in a stream where order matters.

try {
    $this->handle($message);
} catch (UnprocessableMessage $e) {
    $this->alerts->critical('consumer halted', [
        'partition' => $message->partition,
        'offset'    => $message->offset,
        'reason'    => $e->getMessage(),
    ]);

    return self::HALT;   // do not advance the offset
}

Halting is the right answer when skipping would corrupt the state that follows, and it is a decision to make per stream rather than globally — most queues should dead-letter and continue. What it requires is an alert that genuinely wakes somebody, because a halted consumer is an outage with a delay, and a runbook that covers both fixing forward and skipping deliberately.