pd_client: fix the kvproto compatibility

components/error_code/src/pd.rs

@@ -12,5 +12,6 @@ define_error_codes!(
     REGION_NOT_FOUND => ("RegionNotFound", "", ""),
     STORE_TOMBSTONE => ("StoreTombstone", "", ""),
     GLOBAL_CONFIG_NOT_FOUND => ("GlobalConfigNotFound","",""),
+    DATA_COMPACTED => ("DataCompacted","",""),
     UNKNOWN => ("Unknown", "", "")
 );

components/pd_client/src/client.rs

@@ -302,11 +302,7 @@ impl PdClient for RpcClient {
                     Ok(grpc_response) => {
                         let mut res = HashMap::with_capacity(grpc_response.get_items().len());
                         for c in grpc_response.get_items() {
-                            if c.has_error() {
-                                error!("failed to load global config with key {:?}", c.get_error());
-                            } else {
-                                res.insert(c.get_name().to_owned(), c.get_value().to_owned());
-                            }
+                            res.insert(c.get_name().to_owned(), c.get_value().to_owned());
                         }
                         Ok(res)
                     }

components/pd_client/src/client_v2.rs

@@ -803,11 +803,7 @@ impl PdClient for RpcClient {
                 Ok(grpc_response) => {
                     let mut res = HashMap::with_capacity(grpc_response.get_items().len());
                     for c in grpc_response.get_items() {
-                        if c.has_error() {
-                            error!("failed to load global config with key {:?}", c.get_error());
-                        } else {
-                            res.insert(c.get_name().to_owned(), c.get_value().to_owned());
-                        }
+                        res.insert(c.get_name().to_owned(), c.get_value().to_owned());
                     }
                     Ok(res)
                 }

components/pd_client/src/errors.rs

@@ -26,6 +26,8 @@ pub enum Error {
     StoreTombstone(String),
     #[error("global config item {0} not found")]
     GlobalConfigNotFound(String),
+    #[error("required watch revision is smaller than current compact/min revision. {0:?}")]
+    DataCompacted(String),
 }
 
 pub type Result<T> = result::Result<T, Error>;
@@ -38,6 +40,7 @@ impl Error {
             | Error::RegionNotFound(_)
             | Error::StoreTombstone(_)
             | Error::GlobalConfigNotFound(_)
+            | Error::DataCompacted(_)
             | Error::ClusterBootstrapped(_)
             | Error::Incompatible => false,
         }
@@ -55,6 +58,7 @@ impl ErrorCodeExt for Error {
             Error::RegionNotFound(_) => error_code::pd::REGION_NOT_FOUND,
             Error::StoreTombstone(_) => error_code::pd::STORE_TOMBSTONE,
             Error::GlobalConfigNotFound(_) => error_code::pd::GLOBAL_CONFIG_NOT_FOUND,
+            Error::DataCompacted(_) => error_code::pd::DATA_COMPACTED,
             Error::Other(_) => error_code::pd::UNKNOWN,
         }
     }

components/pd_client/src/util.rs

@@ -873,6 +873,7 @@ pub fn check_resp_header(header: &ResponseHeader) -> Result<()> {
         ErrorType::GlobalConfigNotFound => {
             Err(Error::GlobalConfigNotFound(err.get_message().to_owned()))
         }
+        ErrorType::DataCompacted => Err(Error::DataCompacted(err.get_message().to_owned())),
         ErrorType::Ok => Ok(()),
         ErrorType::DuplicatedEntry | ErrorType::EntryNotFound => Err(box_err!(err.get_message())),
         ErrorType::Unknown => Err(box_err!(err.get_message())),

components/resource_control/src/resource_group.rs

@@ -51,12 +51,12 @@ impl ResourceGroupManager {
             // TODO: currently we only consider the cpu usage in the read path, we may also take
             // io read bytes into account later.
             (GroupMode::RawMode, true) => rg
-                .get_resource_settings()
+                .get_raw_resource_settings()
                 .get_cpu()
                 .get_settings()
                 .get_fill_rate(),
             (GroupMode::RawMode, false) => rg
-                .get_resource_settings()
+                .get_raw_resource_settings()
                 .get_io_write()
                 .get_settings()
                 .get_fill_rate(),
@@ -327,7 +327,7 @@ mod tests {
                 .set_fill_rate(write_tokens);
             group.set_r_u_settings(ru_setting);
         } else {
-            let mut resource_setting = GroupResourceSettings::new();
+            let mut resource_setting = GroupRawResourceSettings::new();
             resource_setting
                 .mut_cpu()
                 .mut_settings()
@@ -336,7 +336,7 @@ mod tests {
                 .mut_io_write()
                 .mut_settings()
                 .set_fill_rate(write_tokens);
-            group.set_resource_settings(resource_setting);
+            group.set_raw_resource_settings(resource_setting);
         }
         group
     }

etc/error_code.toml

@@ -263,6 +263,11 @@ error = '''
 KV:Pd:GlobalConfigNotFound
 '''
 
+["KV:Pd:DataCompacted"]
+error = '''
+KV:Pd:DataCompacted
+'''
+
 ["KV:Pd:Unknown"]
 error = '''
 KV:Pd:Unknown