Panic in state backend can stall checkpointing

[mwylde]

This code in the parquet state backend:

arroyo/arroyo-state/src/parquet.rs

Lines 1254 to 1260 in a7b9099

	tokio::spawn(async move {
	loop {
	if !self.flush_iteration().await.unwrap() {
	return;
	}
	}
	});

is responsible for flushing data to the filesystem. Currently it unwraps on failure, which causes the flusher to stall, breaking checkpointing:

ERROR arroyo_server_common: panicked at /opt/arroyo/src/arroyo-state/src/parquet.rs:1256:50:
called `Result::unwrap()` on an `Err` value: object store error: Generic { store: "S3", source: Error { retries: 0, message: "request error", source: Some(reqwest::Error { kind: Request, url: Url { scheme: "https", cannot_be_a_base: false, username: "", password: None, host: Some(Domain("s3.ap-southeast-1.amazonaws.com")), port: None, path: "/arroyo-streaming/checkpoints/job_Qwj7ye3Egl/checkpoints/checkpoint-0013414/operator-job_source_0/table-k-000", query: None, fragment: None }, source: hyper::Error(Io, Os { code: 104, kind: ConnectionReset, message: "Connection reset by peer" }) }), status: None } }

Thanks @harshit2283 for the report

[mwylde]

Fixed with #445