fix(services/webhdfs): encode startAfter and abort the blocks that were written

core/services/webhdfs/src/core.rs

@@ -314,7 +314,7 @@ impl WebhdfsCore {
             percent_encode_path(&p),
         );
         if !start_after.is_empty() {
-            url += format!("&startAfter={start_after}").as_str();
+            url += format!("&startAfter={}", percent_encode_path(start_after)).as_str();
         }
         if let Some(user) = &self.user_name {
             url += format!("&user.name={user}").as_str();

core/services/webhdfs/src/writer.rs

@@ -135,10 +135,16 @@ impl oio::BlockWrite for WebhdfsWriter {
     }
 
     async fn abort_block(&self, block_ids: Vec<Uuid>) -> Result<()> {
+        let Some(ref atomic_write_dir) = self.core.atomic_write_dir else {
+            return Err(Error::new(
+                ErrorKind::Unsupported,
+                "write multi is not supported when atomic is not set",
+            ));
+        };
         for block_id in block_ids {
             let resp = self
                 .core
-                .webhdfs_delete(&self.ctx, &block_id.to_string())
+                .webhdfs_delete(&self.ctx, &format!("{atomic_write_dir}{block_id}"))
                 .await?;
             match resp.status() {
                 StatusCode::OK => {}