fix: panic on unexpected filename structure

Author: Devdutt Shenoi

src/parseable/streams.rs

@@ -70,6 +70,20 @@ static ARROWS_NAME_STRUCTURE: Lazy<Regex> = Lazy::new(|| {
     Regex::new(r"^[[:alnum:]]+\.(?P<front>\S+)\.\d+\.data\.arrows$").expect("Validated regex")
 });
 
+fn arrow_path_to_parquet(path: &Path, random_string: &str) -> Option<PathBuf> {
+    let filename = path.file_name().unwrap().to_str().unwrap();
+    let filename = ARROWS_NAME_STRUCTURE
+        .captures(filename)
+        .and_then(|c| c.get(1))?
+        .as_str();
+    let filename_with_random_number = format!("{filename}.data.{random_string}.arrows");
+    let mut parquet_path = path.to_owned();
+    parquet_path.set_file_name(filename_with_random_number);
+    parquet_path.set_extension("parquet");
+
+    Some(parquet_path)
+}
+
 #[derive(Debug, thiserror::Error)]
 #[error("Stream not found: {0}")]
 pub struct StreamNotFound(pub String);
@@ -222,12 +236,13 @@ impl Stream {
                     &arrow_file_path, self.stream_name
                 );
                 remove_file(&arrow_file_path).unwrap();
-            } else {
-                let key = Self::arrow_path_to_parquet(&arrow_file_path, &random_string);
+            } else if let Some(key) = arrow_path_to_parquet(&arrow_file_path, &random_string) {
                 grouped_arrow_file
                     .entry(key)
                     .or_default()
                     .push(arrow_file_path);
+            } else {
+                warn!("Unexpected arrows file: {}", arrow_file_path.display());
             }
         }
         grouped_arrow_file
@@ -286,21 +301,6 @@ impl Stream {
         }
     }
 
-    fn arrow_path_to_parquet(path: &Path, random_string: &str) -> PathBuf {
-        let filename = path.file_name().unwrap().to_str().unwrap();
-        let filename = ARROWS_NAME_STRUCTURE
-            .captures(filename)
-            .unwrap()
-            .get(1)
-            .unwrap()
-            .as_str();
-        let filename_with_random_number = format!("{filename}.data.{random_string}.arrows");
-        let mut parquet_path = path.to_owned();
-        parquet_path.set_file_name(filename_with_random_number);
-        parquet_path.set_extension("parquet");
-        parquet_path
-    }
-
     /// Converts arrow files in staging into parquet files, does so only for past minutes when run with `!shutdown_signal`
     pub fn prepare_parquet(&self, shutdown_signal: bool) -> Result<(), StagingError> {
         info!(