Merge pull request #9227 from drmingdrmer/30-fixup

refactor(meta): remove back compatability supporting code

Author: BohuTANG

src/meta/client/src/grpc_action.rs

@@ -45,20 +45,6 @@ pub trait RequestFor {
     type Reply;
 }
 
-// TODO: reduce this and MetaGrpcReadReq into one enum?
-// Action wrapper for do_action.
-#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, derive_more::From)]
-pub enum MetaGrpcWriteReq {
-    UpsertKV(UpsertKVReq),
-}
-
-#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, derive_more::From)]
-pub enum MetaGrpcReadReq {
-    GetKV(GetKVReq),
-    MGetKV(MGetKVReq),
-    ListKV(ListKVReq), // since 2022-05-23
-}
-
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, derive_more::From)]
 pub enum MetaGrpcReq {
     UpsertKV(UpsertKVReq),
@@ -81,15 +67,6 @@ impl TryInto<MetaGrpcReq> for Request<RaftRequest> {
     }
 }
 
-impl tonic::IntoRequest<RaftRequest> for MetaGrpcWriteReq {
-    fn into_request(self) -> Request<RaftRequest> {
-        let raft_request = RaftRequest {
-            data: serde_json::to_string(&self).expect("fail to serialize"),
-        };
-        tonic::Request::new(raft_request)
-    }
-}
-
 impl TryInto<Request<RaftRequest>> for MetaGrpcReq {
     type Error = serde_json::Error;
 
@@ -103,32 +80,6 @@ impl TryInto<Request<RaftRequest>> for MetaGrpcReq {
     }
 }
 
-impl TryInto<Request<RaftRequest>> for MetaGrpcWriteReq {
-    type Error = serde_json::Error;
-
-    fn try_into(self) -> Result<Request<RaftRequest>, Self::Error> {
-        let raft_request = RaftRequest {
-            data: serde_json::to_string(&self)?,
-        };
-
-        let request = tonic::Request::new(raft_request);
-        Ok(request)
-    }
-}
-
-impl TryInto<Request<RaftRequest>> for MetaGrpcReadReq {
-    type Error = serde_json::Error;
-
-    fn try_into(self) -> Result<Request<RaftRequest>, Self::Error> {
-        let get_req = RaftRequest {
-            data: serde_json::to_string(&self)?,
-        };
-
-        let request = tonic::Request::new(get_req);
-        Ok(request)
-    }
-}
-
 impl RequestFor for GetKVReq {
     type Reply = GetKVReply;
 }

src/meta/client/src/grpc_client.rs

@@ -715,21 +715,22 @@ impl MetaGrpcClient {
         let resp =
             res.map_err(|status| MetaHandshakeError::new("handshake is refused", &status))?;
 
-        // backward compatibility: no version in handshake.
-        // TODO(xp): remove this when merged.
-        if resp.protocol_version > 0 {
-            let min_compatible = to_digit_ver(min_metasrv_ver);
-            if resp.protocol_version < min_compatible {
-                let invalid_err = AnyError::error(format!(
-                    "metasrv protocol_version({}) < meta-client min-compatible({})",
-                    from_digit_ver(resp.protocol_version),
-                    min_metasrv_ver,
-                ));
-                return Err(MetaHandshakeError::new(
-                    "incompatible protocol version",
-                    &invalid_err,
-                ));
-            }
+        assert!(
+            resp.protocol_version > 0,
+            "talking to a very old databend-meta: upgrade databend-meta to at least 0.8"
+        );
+
+        let min_compatible = to_digit_ver(min_metasrv_ver);
+        if resp.protocol_version < min_compatible {
+            let invalid_err = AnyError::error(format!(
+                "metasrv protocol_version({}) < meta-client min-compatible({})",
+                from_digit_ver(resp.protocol_version),
+                min_metasrv_ver,
+            ));
+            return Err(MetaHandshakeError::new(
+                "incompatible protocol version",
+                &invalid_err,
+            ));
         }
 
         let token = resp.payload;

src/meta/client/src/lib.rs

@@ -17,9 +17,7 @@ mod grpc_client;
 mod kv_api_impl;
 mod message;
 
-pub use grpc_action::MetaGrpcReadReq;
 pub use grpc_action::MetaGrpcReq;
-pub use grpc_action::MetaGrpcWriteReq;
 pub use grpc_action::RequestFor;
 pub use grpc_client::ClientHandle;
 pub use grpc_client::MetaGrpcClient;

src/meta/client/tests/it/grpc_server.rs

@@ -17,6 +17,8 @@ use std::thread::sleep;
 use std::time::Duration;
 
 use common_base::base::tokio;
+use common_meta_client::to_digit_ver;
+use common_meta_client::MIN_METASRV_SEMVER;
 use common_meta_types::protobuf::meta_service_server::MetaService;
 use common_meta_types::protobuf::meta_service_server::MetaServiceServer;
 use common_meta_types::protobuf::ClientInfo;
@@ -51,7 +53,12 @@ impl MetaService for GrpcServiceForTestImpl {
         _request: Request<Streaming<common_meta_types::protobuf::HandshakeRequest>>,
     ) -> Result<Response<Self::HandshakeStream>, Status> {
         tokio::time::sleep(Duration::from_secs(2)).await;
-        let output = futures::stream::once(async { Ok(HandshakeResponse::default()) });
+        let output = futures::stream::once(async {
+            Ok(HandshakeResponse {
+                protocol_version: to_digit_ver(&MIN_METASRV_SEMVER),
+                payload: vec![],
+            })
+        });
         Ok(Response::new(Box::pin(output)))
     }