chore(deps): Bump OpenDAL to v0.41

.github/actions/spelling/allow.txt

@@ -113,6 +113,7 @@ Mediacom
 Medion
 Meizu
 Mertz
+metakey
 Metakey
 Micromax
 Mito

Cargo.toml

@@ -184,7 +184,7 @@ azure_storage = { version = "0.16", default-features = false, optional = true }
 azure_storage_blobs = { version = "0.16", default-features = false, optional = true }
 
 # OpenDAL
-opendal = {version = "0.38", default-features = false, features = ["native-tls", "services-webhdfs"], optional = true}
+opendal = {version = "0.41", default-features = false, features = ["native-tls", "services-webhdfs"], optional = true}
 
 # Tower
 tower = { version = "0.4.13", default-features = false, features = ["buffer", "limit", "retry", "timeout", "util", "balance", "discover"] }

src/sinks/webhdfs/integration_tests.rs

@@ -3,7 +3,7 @@ use std::{
     time::Duration,
 };
 
-use futures::{stream, StreamExt};
+use futures::stream;
 use opendal::{Entry, Metakey};
 use similar_asserts::assert_eq;
 use vector_lib::codecs::{encoding::FramingConfig, TextSerializerConfig};
@@ -74,25 +74,12 @@ async fn hdfs_rotate_files_after_the_buffer_size_is_reached() {
     // Hard-coded sleeps are bad, but we're waiting on localstack's state to converge.
     tokio::time::sleep(Duration::from_secs(1)).await;
 
-    // blocking_scan isn't supported
-    let objects: Vec<Entry> = op
-        .scan("/")
+    let mut objects: Vec<Entry> = op
+        .list_with("/")
+        .delimiter("")
+        .metakey(Metakey::Mode)
         .await
-        .unwrap()
-        .map(|x| x.unwrap())
-        .collect()
-        .await;
-
-    let mut objects = objects
-        .into_iter()
-        .filter(|entry| {
-            op.blocking()
-                .metadata(entry, Metakey::Mode)
-                .unwrap()
-                .mode()
-                .is_file()
-        })
-        .collect::<Vec<_>>();
+        .unwrap();
 
     // Sort file path in order, because we have the event id in path.
     objects.sort_by(|l, r| l.path().cmp(r.path()));