fix(consumer): Don't Crash Rust Consumer on Transport Error

[george-sentry]

Linear

Completes STREAM-856

Description

Python and Rust Arroyo consumers handle errors differently during poll.

In Python, consumers raise TransportError if they hit KafkaError._TRANSPORT during poll.

arroyo/arroyo/backends/kafka/consumer.py

Lines 501 to 502 in e33f4d6

	elif code == KafkaError._TRANSPORT:
	raise TransportError(str(error))

TransportError is a RecoverableError, so StreamProcessor just swallows the error and goes to the next loop.

arroyo/arroyo/processing/processor.py

Lines 458 to 459 in e33f4d6

	except RecoverableError:
	return

Rust doesn't seem to have a concept of recoverable errors, so it just crashes when we hit any error during poll.

arroyo/rust-arroyo/src/processing/mod.rs

Line 442 in e33f4d6

tracing::info!("Caught error, terminating strategy");

We need to update Rust Arroyo to retry / ignore KafkaError._TRANSPORT. This PR accomplishes that by checking whether the error is one we consider to be "recoverable", and if it is, we return Ok(None). If it isn't one we consider "recoverable," we return an error as before.

[linear-code]

STREAM-856 Rust arroyo consumers crash on transient errors during poll

[george-sentry]

My solution is kind of ad hoc, and doesn't really reflect the Python logic. I could do that by extending the Rust ConsumerError enum with a new member like RecoverableError or TransportError.

[untitaker]

maybe a bad idea, but you can make this a macro like such:

macro_rules recoverable_error_pattern {
    () => ((KafkaError::MessageConsumption(RDKafkaErrorCode::BrokerTransportFailure)
            | KafkaError::Global(RDKafkaErrorCode::BrokerTransportFailure));
}

...

match res {
    Some(Err(recoverable_error_pattern!())) => ...,
    Some(Err(err)) => ...
}

then the exhaustiveness and dead code checks of the compiler will continue to be effective because you avoided one if matches!()