Test new object_store

Cargo.toml

@@ -80,5 +80,7 @@ datafusion-physical-plan = {git = 'https://github.com/ArroyoSystems/arrow-datafu
 datafusion-proto = {git = 'https://github.com/ArroyoSystems/arrow-datafusion', branch = '40.0.0/arroyo'}
 datafusion-functions = {git = 'https://github.com/ArroyoSystems/arrow-datafusion', branch = '40.0.0/arroyo'}
 
+object_store = { git = 'http://github.com/ArroyoSystems/arrow-rs', branch = 'gcp_bugs' }
+
 cornucopia_async = { git = "https://github.com/ArroyoSystems/cornucopia", branch = "sqlite" }
 cornucopia = { git = "https://github.com/ArroyoSystems/cornucopia", branch = "sqlite" }

crates/arroyo-controller/src/job_controller/mod.rs

@@ -759,7 +759,7 @@ impl JobController {
                 self.cleanup_task = Some(self.start_cleanup(new_epoch));
             }
         }
-
+        
         // check on checkpointing
         if self.model.checkpoint_state.is_some() {
             self.model.finish_checkpoint_if_done(&self.db).await?;

crates/arroyo-state/src/parquet.rs

@@ -16,7 +16,6 @@ use std::collections::{HashMap, HashSet};
 use std::ops::RangeInclusive;
 use std::sync::Arc;
 use std::time::SystemTime;
-use tokio::sync::Mutex;
 use tracing::{debug, info};
 
 pub const FULL_KEY_RANGE: RangeInclusive<u64> = 0..=u64::MAX;
@@ -122,7 +121,7 @@ impl BackingStore for ParquetBackend {
             })
             .collect();
 
-        let storage_client = Mutex::new(get_storage_provider().await?);
+        let storage_client = get_storage_provider().await?;
 
         // wait for all of the futures to complete
         while let Some(result) = futures.next().await {
@@ -134,7 +133,7 @@ impl BackingStore for ParquetBackend {
                     epoch_to_remove,
                     &operator_id,
                 ));
-                storage_client.lock().await.delete_if_present(path).await?;
+                storage_client.delete_if_present(path).await?;
             }
             debug!(
                 message = "Finished cleaning operator",
@@ -146,8 +145,6 @@ impl BackingStore for ParquetBackend {
 
         for epoch_to_remove in old_min_epoch..min_epoch {
             storage_client
-                .lock()
-                .await
                 .delete_if_present(metadata_path(&base_path(&metadata.job_id, epoch_to_remove)))
                 .await?;
         }