add integration test for replace dn on failure policy

Author: zhaohaiyuan

rust/tests/test_write_resiliency.rs

@@ -1,7 +1,11 @@
-#[cfg(feature = "integration-test")]
+// #[cfg(feature = "integration-test")]
 mod test {
 
-    use std::{collections::HashSet, sync::atomic::Ordering, time::Duration};
+    use std::{
+        collections::{HashMap, HashSet},
+        sync::atomic::Ordering,
+        time::Duration,
+    };
 
     use bytes::{Buf, BufMut, Bytes, BytesMut};
     use hdfs_native::{
@@ -68,79 +72,104 @@ mod test {
     }
 
     async fn test_write_failures() -> Result<()> {
-        let client = Client::default();
-
-        let file = "/testfile";
-        let bytes_to_write = 2usize * 1024 * 1024;
-
-        let mut data = BytesMut::with_capacity(bytes_to_write);
-        for i in 0..(bytes_to_write / 4) {
-            data.put_i32(i as i32);
+        let mut replace_dn_on_failure_conf: HashMap<String, String> = HashMap::new();
+        replace_dn_on_failure_conf.insert(
+            "dfs.client.block.write.replace-datanode-on-failure.enable".to_string(),
+            "true".to_string(),
+        );
+        replace_dn_on_failure_conf.insert(
+            "dfs.client.block.write.replace-datanode-on-failure.policy".to_string(),
+            "ALWAYS".to_string(),
+        );
+
+        for (i, client) in vec![
+            Client::default(),
+            // Client::default_with_config(replace_dn_on_failure_conf).unwrap(),
+        ]
+        .iter()
+        .enumerate()
+        {
+            let file = format!("/testfile{}", i);
+            let bytes_to_write = 2usize * 1024 * 1024;
+
+            let mut data = BytesMut::with_capacity(bytes_to_write);
+            for i in 0..(bytes_to_write / 4) {
+                data.put_i32(i as i32);
+            }
+
+            // Test connection failure before writing data
+            let mut writer = client
+                .create(&file, WriteOptions::default().replication(3))
+                .await?;
+
+            WRITE_CONNECTION_FAULT_INJECTOR.store(true, Ordering::SeqCst);
+
+            let data = data.freeze();
+            writer.write(data.clone()).await?;
+            writer.close().await?;
+
+            let reader = client.read(&file).await?;
+            check_file_content(&reader, data.clone()).await?;
+
+            // Test connection failure after data has been written
+            let mut writer = client
+                .create(
+                    &file,
+                    WriteOptions::default().replication(3).overwrite(true),
+                )
+                .await?;
+
+            writer.write(data.slice(..bytes_to_write / 2)).await?;
+
+            // Give a little time for the packets to send
+            tokio::time::sleep(Duration::from_millis(100)).await;
+
+            WRITE_CONNECTION_FAULT_INJECTOR.store(true, Ordering::SeqCst);
+
+            writer.write(data.slice(bytes_to_write / 2..)).await?;
+            writer.close().await?;
+
+            let reader = client.read(&file).await?;
+            check_file_content(&reader, data.clone()).await?;
+
+            // Test failure in from ack status before any data is written
+            let mut writer = client
+                .create(
+                    &file,
+                    WriteOptions::default().replication(3).overwrite(true),
+                )
+                .await?;
+
+            *WRITE_REPLY_FAULT_INJECTOR.lock().unwrap() = Some(2);
+
+            writer.write(data.clone()).await?;
+            writer.close().await?;
+
+            let reader = client.read(&file).await?;
+            check_file_content(&reader, data.clone()).await?;
+
+            // Test failure in from ack status after some data has been written
+            let mut writer = client
+                .create(
+                    &file,
+                    WriteOptions::default().replication(3).overwrite(true),
+                )
+                .await?;
+
+            writer.write(data.slice(..bytes_to_write / 2)).await?;
+
+            // Give a little time for the packets to send
+            tokio::time::sleep(Duration::from_millis(100)).await;
+
+            *WRITE_REPLY_FAULT_INJECTOR.lock().unwrap() = Some(2);
+
+            writer.write(data.slice(bytes_to_write / 2..)).await?;
+            writer.close().await?;
+
+            let reader = client.read(&file).await?;
+            check_file_content(&reader, data.clone()).await?;
         }
 
-        // Test connection failure before writing data
-        let mut writer = client
-            .create(file, WriteOptions::default().replication(3))
-            .await?;
-
-        WRITE_CONNECTION_FAULT_INJECTOR.store(true, Ordering::SeqCst);
-
-        let data = data.freeze();
-        writer.write(data.clone()).await?;
-        writer.close().await?;
-
-        let reader = client.read("/testfile").await?;
-        check_file_content(&reader, data.clone()).await?;
-
-        // Test connection failure after data has been written
-        let mut writer = client
-            .create(file, WriteOptions::default().replication(3).overwrite(true))
-            .await?;
-
-        writer.write(data.slice(..bytes_to_write / 2)).await?;
-
-        // Give a little time for the packets to send
-        tokio::time::sleep(Duration::from_millis(100)).await;
-
-        WRITE_CONNECTION_FAULT_INJECTOR.store(true, Ordering::SeqCst);
-
-        writer.write(data.slice(bytes_to_write / 2..)).await?;
-        writer.close().await?;
-
-        let reader = client.read("/testfile").await?;
-        check_file_content(&reader, data.clone()).await?;
-
-        // Test failure in from ack status before any data is written
-        let mut writer = client
-            .create(file, WriteOptions::default().replication(3).overwrite(true))
-            .await?;
-
-        *WRITE_REPLY_FAULT_INJECTOR.lock().unwrap() = Some(2);
-
-        writer.write(data.clone()).await?;
-        writer.close().await?;
-
-        let reader = client.read("/testfile").await?;
-        check_file_content(&reader, data.clone()).await?;
-
-        // Test failure in from ack status after some data has been written
-        let mut writer = client
-            .create(file, WriteOptions::default().replication(3).overwrite(true))
-            .await?;
-
-        writer.write(data.slice(..bytes_to_write / 2)).await?;
-
-        // Give a little time for the packets to send
-        tokio::time::sleep(Duration::from_millis(100)).await;
-
-        *WRITE_REPLY_FAULT_INJECTOR.lock().unwrap() = Some(2);
-
-        writer.write(data.slice(bytes_to_write / 2..)).await?;
-        writer.close().await?;
-
-        let reader = client.read("/testfile").await?;
-        check_file_content(&reader, data.clone()).await?;
-
         Ok(())
     }