fix(hdfs): Append another retry file when appending complete failed

Author: zuston

src/config.rs

@@ -66,7 +66,7 @@ impl MemoryStoreConfig {
 
 // =========================================================
 
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Default)]
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
 pub struct HdfsStoreConfig {
     #[serde(default = "as_default_max_concurrency")]
     pub max_concurrency: usize,
@@ -82,6 +82,16 @@ fn as_default_partition_write_max_concurrency() -> usize {
     20
 }
 
+impl Default for HdfsStoreConfig {
+    fn default() -> Self {
+        Self {
+            max_concurrency: as_default_max_concurrency(),
+            partition_write_max_concurrency: as_default_partition_write_max_concurrency(),
+            kerberos_security_config: None,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Default)]
 pub struct KerberosSecurityConfig {
     pub keytab_path: String,

src/semaphore_with_index.rs

@@ -24,7 +24,7 @@ impl<'a> SemaphorePermitWithIndex<'a> {
 impl<'a> Drop for SemaphorePermitWithIndex<'a> {
     fn drop(&mut self) {
         let mut index_container = self.index_ref.lock();
-        index_container.push_back(self.index);
+        index_container.push_front(self.index);
     }
 }
 
@@ -59,7 +59,6 @@ impl SemaphoreWithIndex {
 #[cfg(test)]
 mod test {
     use crate::semaphore_with_index::SemaphoreWithIndex;
-    use std::sync::atomic::Ordering::SeqCst;
 
     #[tokio::test]
     async fn test() -> anyhow::Result<()> {