ref(consumer): Add batch write timeout to Rust consumer

[ayirr7]

Adds a new CLI arg to the Rust consumers, batch_write_timeout_ms. This parameter ends up in the batch writer (see batch.rs) where the timeout gets applied to the HTTP client that writes data to Clickhouse.

The timeout would get applied to the entire time period of the connection + the time to receive a response from the server.

The reason behind adding this timeout is to give us a configurable way to control the period of time that a consumer could potentially stall after encountering transients Clickhouse connection errors. We have a current theory that such errors can affect specific partitions of a consumer, leading to slower-than-ideal recovery and a backlog on said partitions. Introducing a timeout could allow the partitions to recover more quickly and possibly prevent from such backlogs forming in the first place.

[ayirr7]

I saw some other options for timeout, e.g. https://docs.rs/reqwest/latest/reqwest/struct.ClientBuilder.html#method.connect_timeout. Might be worth double checking which one we want to use?

[untitaker]

this one is fine. it applies to the entire lifecycle of the request.

however, I just realized, it also means that the timeout always needs to be higher than the max batch time...

[ayirr7]

it also means that the timeout always needs to be higher than the max batch time...

Why is this the case? From https://github.com/getsentry/snuba/blob/master/rust_snuba/src/factory.rs#L130-L149 I thought that the batches are built first (based on either batch time/size) and then packaged as a request to Clickhouse? The place where we are applying the timeout is after the batch has already been built and was closed based on batch time/size, no?

[untitaker]

the timeout is for the entire lifecycle of the request, which is from when the batch is being created (=when the http request is being started and starts streaming) to when the batch is flushed (=when the last chunk is being written and we wait for and check the HTTP status code)

[ayirr7]

Added a constraint in consumer.rs to account for this. It makes sure that batch timeout >= batch time, if it is not, then the batch timeout passed in is silently discarded.

[ayirr7]

Tested locally by starting up a consumer with --batch-write-timeout-ms=0 and sending a message, which triggered the timeout and caused the consumer to crash due to panic with the following error messages:

{"timestamp":"2024-06-25T09:00:14.577750Z","level":"INFO","fields":{"message":"Caught error, terminating strategy"},"target":"rust_arroyo::processing"}
{"timestamp":"2024-06-25T09:00:14.578528Z","level":"INFO","fields":{"message":"Partitions to revoke: [Partition { topic: Topic(\"snuba-generic-metrics\"), index: 0 }]"},"target":"rust_arroyo::processing"}
{"timestamp":"2024-06-25T09:00:14.578653Z","level":"INFO","fields":{"message":"Joining strategy"},"target":"rust_arroyo::processing"}
{"timestamp":"2024-06-25T09:00:14.578771Z","level":"WARN","fields":{"message":"Timeout Some(0ns) reached while waiting for tasks to finish"},"target":"rust_arroyo::processing::strategies::reduce"}
{"timestamp":"2024-06-25T09:00:14.580411Z","level":"INFO","fields":{"message":"Partition revocation complete."},"target":"rust_arroyo::processing"}
{"timestamp":"2024-06-25T09:00:14.849284Z","level":"ERROR","fields":{"message":"Strategy(error sending request for url (http://127.0.0.1:8123/?load_balancing=in_order&insert_distributed_sync=1&query=INSERT+INTO+generic_metric_distributions_raw_local+FORMAT+JSONEachRow): operation timed out\n\nCaused by:\n    operation timed out\n\nStack backtrace:\n   0: std::backtrace_rs::backtrace::libunwind::trace\n             at /rustc/a28077b28a02b92985b3a3faecf92813155f1ea1/library/std/src/ ......

@untitaker If I understand correctly, this would be the desired behavior? Locally, the panic also puts the consumer in a bad state, where if I try to start it again, it crashes again without me sending any data through (but this may just be due to some persistent local state). I modified the unit test as well to reflect that this panic would occur.

Also tested that:

• Running consumers without the CLI arg works fine
• Running consumers with a batch timeout lower than a configured batch time does not take effect, consumer operates as it normally does

[untitaker]

it crashes again without me sending any data through (but this may just be due to some persistent local state

it may have some unprocessed messages from the previous crash, as it didn't commit offsets

anyway, this behavior is fine, it now doesn't hang anymore and that's certainly an improvement with regard to backlog size