strip file:// off of the final_dir for local sinks. (#607)

ArroyoSystems · May 1, 2024 · 3577f04 · 3577f04
1 parent 532d056
commit 3577f04
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 2 deletions.
diff --git a/crates/arroyo-connectors/src/filesystem/sink/local.rs b/crates/arroyo-connectors/src/filesystem/sink/local.rs
@@ -46,10 +46,13 @@ pub struct LocalFileSystemWriter<V: LocalWriter> {
 
 impl<V: LocalWriter> LocalFileSystemWriter<V> {
     pub fn new(
-        final_dir: String,
+        mut final_dir: String,
         table_properties: FileSystemTable,
         config: OperatorConfig,
     ) -> TwoPhaseCommitterOperator<Self> {
+        if final_dir.starts_with("file://") {
+            final_dir = final_dir.trim_start_matches("file://").to_string();
+        }
         // TODO: explore configuration options here
         let tmp_dir = format!("{}/__in_progress", final_dir);
         // make sure final_dir and tmp_dir exists

diff --git a/crates/arroyo-controller/src/states/scheduling.rs b/crates/arroyo-controller/src/states/scheduling.rs
@@ -423,7 +423,7 @@ impl State for Scheduling {
                             arroyo_rpc::grpc::TableEnum::GlobalKeyValue => {
                                 GlobalKeyedTable::committing_data(config.clone(), table_metadata)
                             }
-                            arroyo_rpc::grpc::TableEnum::ExpiringKeyedTimeTable => todo!(),
+                            arroyo_rpc::grpc::TableEnum::ExpiringKeyedTimeTable => None,
                         } {
                             committing_data
                                 .entry(operator_id.clone())