Address tests

Author: Kimahriman

rust/minidfs/src/main/java/main/Main.java

@@ -136,7 +136,7 @@ public static void main(String args[]) throws Exception {
                 dfs.transitionToActive(2);
             } else if (flags.contains("ha")) {
                 activeNamenode = 2;
-                // dfs.transitionToObserver(1);
+                dfs.transitionToObserver(1);
                 dfs.transitionToActive(activeNamenode);
             }
 

rust/src/hdfs/proxy.rs

@@ -114,7 +114,7 @@ impl NameServiceProxy {
     async fn msync_if_needed(&self, write: bool) -> Result<()> {
         if !self.msycned.fetch_or(true, Ordering::SeqCst) && !write {
             let msync_msg = hdfs::MsyncRequestProto::default();
-            self.call_inner("msync", msync_msg.encode_length_delimited_to_vec(), false)
+            self.call_inner("msync", msync_msg.encode_length_delimited_to_vec(), true)
                 .await
                 .map(|_| ())
                 .or_else(|err| match err {
@@ -196,17 +196,23 @@ impl NameServiceProxy {
                 }
                 Err(_) if attempts >= self.proxy_connections.len() - 1 => return result,
                 // Retriable error, do nothing and try the next connection
-                Err(HdfsError::RPCError(exception, _)) => match exception.as_ref() {
-                    OBSERVER_RETRY_EXCEPTION => {
-                        self.current_observers.lock().unwrap().insert(proxy_index);
-                    }
-                    STANDBY_EXCEPTION => {
-                        self.current_observers.lock().unwrap().remove(&proxy_index);
+                Err(HdfsError::RPCError(exception, _))
+                | Err(HdfsError::FatalRPCError(exception, _))
+                    if Self::is_retriable(&exception) =>
+                {
+                    match exception.as_ref() {
+                        OBSERVER_RETRY_EXCEPTION => {
+                            self.current_observers.lock().unwrap().insert(proxy_index);
+                        }
+                        STANDBY_EXCEPTION => {
+                            self.current_observers.lock().unwrap().remove(&proxy_index);
+                        }
+                        _ => (),
                     }
-                    _ => (),
-                },
+                }
                 Err(e) => {
                     // Some other error, we will retry but log the error
+                    self.current_observers.lock().unwrap().remove(&proxy_index);
                     warn!("{:?}", e);
                 }
             }

rust/src/security/sasl.rs

@@ -1,6 +1,6 @@
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 use cipher::{KeyIvInit, StreamCipher};
-use log::{debug, warn};
+use log::debug;
 use prost::Message;
 use std::io;
 use std::sync::{Arc, Mutex};
@@ -262,11 +262,6 @@ impl SaslReader {
                 ));
             }
             RpcStatusProto::Fatal => {
-                warn!(
-                    "RPC fatal error: {}: {}",
-                    rpc_response.exception_class_name(),
-                    rpc_response.error_msg()
-                );
                 return Err(HdfsError::FatalRPCError(
                     rpc_response.exception_class_name().to_string(),
                     rpc_response.error_msg().to_string(),