This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss-rust.git
The following commit(s) were added to refs/heads/main by this push:
new dbbaf9b8 [rust] Support API_VERSION in rust client. (#531)
dbbaf9b8 is described below
commit dbbaf9b8c398d019cef3bca1065a83e8d5eaea0a
Author: Hongshun Wang <[email protected]>
AuthorDate: Sat May 9 16:53:13 2026 +0800
[rust] Support API_VERSION in rust client. (#531)
* [rust] Support API_VERSION in rust client
* remove useless write_version_type and read_version_type which is same as
java client does.
* Add test about kv_format_v2.
* return apiKey.highestSupportedVersion as default.
* check whether server type mismatch.
* add test about unknown server type
---
crates/fluss-test-cluster/test-images.env | 2 +-
crates/fluss/src/client/lookup/lookup_sender.rs | 8 +-
crates/fluss/src/cluster/mod.rs | 17 +
crates/fluss/src/error.rs | 12 +
crates/fluss/src/proto/fluss_api.proto | 19 +-
crates/fluss/src/rpc/api_key.rs | 45 +++
.../message/{init_writer.rs => api_versions.rs} | 38 +--
crates/fluss/src/rpc/message/authenticate.rs | 10 +-
crates/fluss/src/rpc/message/create_database.rs | 11 +-
crates/fluss/src/rpc/message/create_partition.rs | 11 +-
crates/fluss/src/rpc/message/create_table.rs | 11 +-
crates/fluss/src/rpc/message/database_exists.rs | 10 +-
crates/fluss/src/rpc/message/drop_database.rs | 10 +-
crates/fluss/src/rpc/message/drop_partition.rs | 11 +-
crates/fluss/src/rpc/message/drop_table.rs | 11 +-
crates/fluss/src/rpc/message/fetch.rs | 11 +-
crates/fluss/src/rpc/message/get_database_info.rs | 10 +-
.../src/rpc/message/get_latest_lake_snapshot.rs | 10 +-
crates/fluss/src/rpc/message/get_security_token.rs | 10 +-
crates/fluss/src/rpc/message/get_table.rs | 10 +-
crates/fluss/src/rpc/message/get_table_schema.rs | 10 +-
crates/fluss/src/rpc/message/header.rs | 10 +-
crates/fluss/src/rpc/message/init_writer.rs | 11 +-
crates/fluss/src/rpc/message/limit_scan.rs | 11 +-
crates/fluss/src/rpc/message/list_databases.rs | 10 +-
crates/fluss/src/rpc/message/list_offsets.rs | 13 +-
.../fluss/src/rpc/message/list_partition_infos.rs | 11 +-
crates/fluss/src/rpc/message/list_tables.rs | 11 +-
crates/fluss/src/rpc/message/lookup.rs | 13 +-
crates/fluss/src/rpc/message/mod.rs | 31 +-
crates/fluss/src/rpc/message/prefix_lookup.rs | 13 +-
crates/fluss/src/rpc/message/produce_log.rs | 11 +-
crates/fluss/src/rpc/message/put_kv.rs | 11 +-
crates/fluss/src/rpc/message/table_exists.rs | 11 +-
crates/fluss/src/rpc/message/update_metadata.rs | 11 +-
crates/fluss/src/rpc/server_connection.rs | 341 ++++++++++++++++++---
crates/fluss/tests/integration/kv_table.rs | 71 +++++
37 files changed, 611 insertions(+), 266 deletions(-)
diff --git a/crates/fluss-test-cluster/test-images.env
b/crates/fluss-test-cluster/test-images.env
index 3aa8e735..5cd91417 100644
--- a/crates/fluss-test-cluster/test-images.env
+++ b/crates/fluss-test-cluster/test-images.env
@@ -1,4 +1,4 @@
FLUSS_IMAGE=apache/fluss
-FLUSS_VERSION=0.9.0-incubating
+FLUSS_VERSION=0.9.1-incubating
ZOOKEEPER_IMAGE=zookeeper
ZOOKEEPER_VERSION=3.9.2
diff --git a/crates/fluss/src/client/lookup/lookup_sender.rs
b/crates/fluss/src/client/lookup/lookup_sender.rs
index efcd6853..1e337f83 100644
--- a/crates/fluss/src/client/lookup/lookup_sender.rs
+++ b/crates/fluss/src/client/lookup/lookup_sender.rs
@@ -22,9 +22,7 @@ use crate::error::{Error, FlussError, Result};
use crate::metadata::{TableBucket, TablePath};
use crate::proto::{LookupResponse, PrefixLookupResponse};
use crate::rpc::ServerConnection;
-use crate::rpc::message::{
- LookupRequest, PrefixLookupRequest, ReadVersionedType, RequestBody,
WriteVersionedType,
-};
+use crate::rpc::message::{LookupRequest, PrefixLookupRequest, ReadType,
RequestBody, WriteType};
use crate::{BucketId, PartitionId, TableId};
use bytes::Bytes;
use futures::stream::{FuturesUnordered, StreamExt};
@@ -51,8 +49,8 @@ struct BucketResponse<V> {
}
trait LookupProtocol {
- type Request: RequestBody<ResponseBody = Self::Response> + Send +
WriteVersionedType<Vec<u8>>;
- type Response: ReadVersionedType<Cursor<Vec<u8>>> + Send;
+ type Request: RequestBody<ResponseBody = Self::Response> + Send +
WriteType<Vec<u8>>;
+ type Response: ReadType<Cursor<Vec<u8>>> + Send;
type Value: Send;
const OP_NAME: &'static str;
diff --git a/crates/fluss/src/cluster/mod.rs b/crates/fluss/src/cluster/mod.rs
index 8b825eee..863f8ed5 100644
--- a/crates/fluss/src/cluster/mod.rs
+++ b/crates/fluss/src/cluster/mod.rs
@@ -79,6 +79,23 @@ pub enum ServerType {
CoordinatorServer,
}
+impl ServerType {
+ pub fn to_type_id(&self) -> i32 {
+ match self {
+ ServerType::CoordinatorServer => 1,
+ ServerType::TabletServer => 2,
+ }
+ }
+
+ pub fn from_type_id(type_id: i32) -> Option<ServerType> {
+ match type_id {
+ 1 => Some(ServerType::CoordinatorServer),
+ 2 => Some(ServerType::TabletServer),
+ _ => None,
+ }
+ }
+}
+
impl fmt::Display for ServerType {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
diff --git a/crates/fluss/src/error.rs b/crates/fluss/src/error.rs
index c2f72a74..4bd0690e 100644
--- a/crates/fluss/src/error.rs
+++ b/crates/fluss/src/error.rs
@@ -115,6 +115,18 @@ pub enum Error {
#[snafu(visibility(pub(crate)), display("Fluss API Error: {}.",
api_error))]
FlussAPIError { api_error: ApiError },
+
+ #[snafu(
+ visibility(pub(crate)),
+ display("Unsupported API version: {}.", message)
+ )]
+ UnsupportedVersion { message: String },
+
+ /// The server advertised a `server_type` that does not match the one
expected
+ /// for the target `ServerNode` (e.g. connecting to a coordinator on a
tablet
+ /// server address).
+ #[snafu(visibility(pub(crate)), display("Invalid server type: {}.",
message))]
+ InvalidServerType { message: String },
}
/// Convenience constructors for API errors that may be raised client-side.
diff --git a/crates/fluss/src/proto/fluss_api.proto
b/crates/fluss/src/proto/fluss_api.proto
index a544906b..2add80d7 100644
--- a/crates/fluss/src/proto/fluss_api.proto
+++ b/crates/fluss/src/proto/fluss_api.proto
@@ -24,6 +24,23 @@ message ErrorResponse {
optional string error_message = 2;
}
+// api versions request and response
+message ApiVersionsRequest {
+ required string client_software_name = 1;
+ required string client_software_version = 2;
+}
+
+message ApiVersionsResponse {
+ repeated PbApiVersion api_versions = 1;
+ optional int32 server_type = 2;
+}
+
+message PbApiVersion {
+ required int32 api_key = 1;
+ required int32 min_version = 2;
+ required int32 max_version = 3;
+}
+
// metadata request and response, request send from client to each server.
message MetadataRequest {
repeated PbTablePath table_path = 1;
@@ -483,4 +500,4 @@ message InitWriterRequest {
message InitWriterResponse {
required int64 writer_id = 1;
-}
\ No newline at end of file
+}
diff --git a/crates/fluss/src/rpc/api_key.rs b/crates/fluss/src/rpc/api_key.rs
index 1ea0269d..977b69d1 100644
--- a/crates/fluss/src/rpc/api_key.rs
+++ b/crates/fluss/src/rpc/api_key.rs
@@ -16,9 +16,11 @@
// under the License.
use crate::rpc::api_key::ApiKey::Unknown;
+use crate::rpc::api_version::{ApiVersion, ApiVersionRange};
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy)]
pub enum ApiKey {
+ ApiVersion, // 1000
CreateDatabase, // 1001
DropDatabase, // 1002
ListDatabases, // 1003
@@ -48,9 +50,50 @@ pub enum ApiKey {
Unknown(i16),
}
+impl ApiKey {
+ /// Returns the range of versions supported by the client for this API key.
+ pub fn supported_versions(&self) -> Option<ApiVersionRange> {
+ match self {
+ // Most APIs only support v0.
+ ApiKey::ApiVersion
+ | ApiKey::CreateDatabase
+ | ApiKey::DropDatabase
+ | ApiKey::ListDatabases
+ | ApiKey::DatabaseExists
+ | ApiKey::CreateTable
+ | ApiKey::DropTable
+ | ApiKey::GetTable
+ | ApiKey::ListTables
+ | ApiKey::ListPartitionInfos
+ | ApiKey::TableExists
+ | ApiKey::GetTableSchema
+ | ApiKey::MetaData
+ | ApiKey::ProduceLog
+ | ApiKey::FetchLog
+ | ApiKey::ListOffsets
+ | ApiKey::GetFileSystemSecurityToken
+ | ApiKey::InitWriter
+ | ApiKey::GetLatestLakeSnapshot
+ | ApiKey::LimitScan
+ | ApiKey::GetDatabaseInfo
+ | ApiKey::CreatePartition
+ | ApiKey::DropPartition
+ | ApiKey::Authenticate
+ // TODO(key-encoding-v1): The Java server supports v0..v1 for these
+ // APIs, but the Rust client has not yet implemented the v1 key
+ // encoding format. Pinned to v0 until that is done.
+ | ApiKey::PutKv | ApiKey::Lookup | ApiKey::PrefixLookup => {
+ Some(ApiVersionRange::new(ApiVersion(0), ApiVersion(0)))
+ }
+ Unknown(_) => None,
+ }
+ }
+}
+
impl From<i16> for ApiKey {
fn from(key: i16) -> Self {
match key {
+ 1000 => ApiKey::ApiVersion,
1001 => ApiKey::CreateDatabase,
1002 => ApiKey::DropDatabase,
1003 => ApiKey::ListDatabases,
@@ -86,6 +129,7 @@ impl From<i16> for ApiKey {
impl From<ApiKey> for i16 {
fn from(key: ApiKey) -> Self {
match key {
+ ApiKey::ApiVersion => 1000,
ApiKey::CreateDatabase => 1001,
ApiKey::DropDatabase => 1002,
ApiKey::ListDatabases => 1003,
@@ -124,6 +168,7 @@ mod tests {
#[test]
fn api_key_round_trip() {
let cases = [
+ (1000, ApiKey::ApiVersion),
(1001, ApiKey::CreateDatabase),
(1002, ApiKey::DropDatabase),
(1003, ApiKey::ListDatabases),
diff --git a/crates/fluss/src/rpc/message/init_writer.rs
b/crates/fluss/src/rpc/message/api_versions.rs
similarity index 53%
copy from crates/fluss/src/rpc/message/init_writer.rs
copy to crates/fluss/src/rpc/message/api_versions.rs
index 0bbb0dc5..579c66a7 100644
--- a/crates/fluss/src/rpc/message/init_writer.rs
+++ b/crates/fluss/src/rpc/message/api_versions.rs
@@ -15,36 +15,36 @@
// specific language governing permissions and limitations
// under the License.
-use crate::proto::{InitWriterResponse, PbTablePath};
+use crate::proto::{
+ ApiVersionsRequest as ProtoApiVersionsRequest, ApiVersionsResponse as
ProtoApiVersionsResponse,
+};
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::{ReadError, WriteError};
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type};
use bytes::{Buf, BufMut};
use prost::Message;
-pub struct InitWriterRequest {
- pub inner_request: proto::InitWriterRequest,
+#[derive(Debug, Clone)]
+pub struct ApiVersionsRequest {
+ pub inner_request: ProtoApiVersionsRequest,
}
-impl InitWriterRequest {
- pub fn new(table_paths: Vec<PbTablePath>) -> Self {
- InitWriterRequest {
- inner_request: proto::InitWriterRequest {
- table_path: table_paths,
+impl ApiVersionsRequest {
+ pub fn new(client_name: &str, client_version: &str) -> Self {
+ Self {
+ inner_request: ProtoApiVersionsRequest {
+ client_software_name: client_name.to_string(),
+ client_software_version: client_version.to_string(),
},
}
}
}
-impl RequestBody for InitWriterRequest {
- type ResponseBody = InitWriterResponse;
-
- const API_KEY: ApiKey = ApiKey::InitWriter;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
+impl RequestBody for ApiVersionsRequest {
+ type ResponseBody = ProtoApiVersionsResponse;
+ const API_KEY: ApiKey = ApiKey::ApiVersion;
}
-impl_write_version_type!(InitWriterRequest);
-impl_read_version_type!(InitWriterResponse);
+impl_write_type!(ApiVersionsRequest);
+impl_read_type!(ProtoApiVersionsResponse);
diff --git a/crates/fluss/src/rpc/message/authenticate.rs
b/crates/fluss/src/rpc/message/authenticate.rs
index 1292cdc9..1874b304 100644
--- a/crates/fluss/src/rpc/message/authenticate.rs
+++ b/crates/fluss/src/rpc/message/authenticate.rs
@@ -17,10 +17,9 @@
use crate::proto::{AuthenticateRequest as ProtoAuthenticateRequest,
AuthenticateResponse};
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::{ReadError, WriteError};
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -61,11 +60,10 @@ impl AuthenticateRequest {
impl RequestBody for AuthenticateRequest {
type ResponseBody = AuthenticateResponse;
const API_KEY: ApiKey = ApiKey::Authenticate;
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(AuthenticateRequest);
-impl_read_version_type!(AuthenticateResponse);
+impl_write_type!(AuthenticateRequest);
+impl_read_type!(AuthenticateResponse);
#[cfg(test)]
mod tests {
diff --git a/crates/fluss/src/rpc/message/create_database.rs
b/crates/fluss/src/rpc/message/create_database.rs
index e03cd1ff..ed0868da 100644
--- a/crates/fluss/src/rpc/message/create_database.rs
+++ b/crates/fluss/src/rpc/message/create_database.rs
@@ -16,15 +16,14 @@
// under the License.
use crate::metadata::DatabaseDescriptor;
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::{impl_read_type, impl_write_type, proto};
use crate::error::Result as FlussResult;
use crate::proto::CreateDatabaseResponse;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::ReadError;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -60,9 +59,7 @@ impl RequestBody for CreateDatabaseRequest {
type ResponseBody = CreateDatabaseResponse;
const API_KEY: ApiKey = ApiKey::CreateDatabase;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(CreateDatabaseRequest);
-impl_read_version_type!(CreateDatabaseResponse);
+impl_write_type!(CreateDatabaseRequest);
+impl_read_type!(CreateDatabaseResponse);
diff --git a/crates/fluss/src/rpc/message/create_partition.rs
b/crates/fluss/src/rpc/message/create_partition.rs
index 93dbf70d..ad633655 100644
--- a/crates/fluss/src/rpc/message/create_partition.rs
+++ b/crates/fluss/src/rpc/message/create_partition.rs
@@ -18,11 +18,10 @@
use crate::metadata::{PartitionSpec, TablePath};
use crate::proto::CreatePartitionResponse;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::convert::to_table_path;
use crate::rpc::frame::{ReadError, WriteError};
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -51,9 +50,7 @@ impl RequestBody for CreatePartitionRequest {
type ResponseBody = CreatePartitionResponse;
const API_KEY: ApiKey = ApiKey::CreatePartition;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(CreatePartitionRequest);
-impl_read_version_type!(CreatePartitionResponse);
+impl_write_type!(CreatePartitionRequest);
+impl_read_type!(CreatePartitionResponse);
diff --git a/crates/fluss/src/rpc/message/create_table.rs
b/crates/fluss/src/rpc/message/create_table.rs
index 69865b89..4647fec6 100644
--- a/crates/fluss/src/rpc/message/create_table.rs
+++ b/crates/fluss/src/rpc/message/create_table.rs
@@ -16,16 +16,15 @@
// under the License.
use crate::metadata::{JsonSerde, TableDescriptor, TablePath};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::{impl_read_type, impl_write_type, proto};
use crate::error::Result as FlussResult;
use crate::proto::CreateTableResponse;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::convert::to_table_path;
use crate::rpc::frame::ReadError;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -55,9 +54,7 @@ impl RequestBody for CreateTableRequest {
type ResponseBody = CreateTableResponse;
const API_KEY: ApiKey = ApiKey::CreateTable;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(CreateTableRequest);
-impl_read_version_type!(CreateTableResponse);
+impl_write_type!(CreateTableRequest);
+impl_read_type!(CreateTableResponse);
diff --git a/crates/fluss/src/rpc/message/database_exists.rs
b/crates/fluss/src/rpc/message/database_exists.rs
index 7e717a4e..4a9588a2 100644
--- a/crates/fluss/src/rpc/message/database_exists.rs
+++ b/crates/fluss/src/rpc/message/database_exists.rs
@@ -18,10 +18,9 @@
use crate::rpc::frame::ReadError;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -44,8 +43,7 @@ impl RequestBody for DatabaseExistsRequest {
type ResponseBody = proto::DatabaseExistsResponse;
const API_KEY: ApiKey = ApiKey::DatabaseExists;
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(DatabaseExistsRequest);
-impl_read_version_type!(proto::DatabaseExistsResponse);
+impl_write_type!(DatabaseExistsRequest);
+impl_read_type!(proto::DatabaseExistsResponse);
diff --git a/crates/fluss/src/rpc/message/drop_database.rs
b/crates/fluss/src/rpc/message/drop_database.rs
index 663e970a..bf7477f3 100644
--- a/crates/fluss/src/rpc/message/drop_database.rs
+++ b/crates/fluss/src/rpc/message/drop_database.rs
@@ -18,10 +18,9 @@
use crate::rpc::frame::ReadError;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -46,8 +45,7 @@ impl RequestBody for DropDatabaseRequest {
type ResponseBody = proto::DropDatabaseResponse;
const API_KEY: ApiKey = ApiKey::DropDatabase;
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(DropDatabaseRequest);
-impl_read_version_type!(proto::DropDatabaseResponse);
+impl_write_type!(DropDatabaseRequest);
+impl_read_type!(proto::DropDatabaseResponse);
diff --git a/crates/fluss/src/rpc/message/drop_partition.rs
b/crates/fluss/src/rpc/message/drop_partition.rs
index ddc97d83..c7494acb 100644
--- a/crates/fluss/src/rpc/message/drop_partition.rs
+++ b/crates/fluss/src/rpc/message/drop_partition.rs
@@ -18,11 +18,10 @@
use crate::metadata::{PartitionSpec, TablePath};
use crate::proto::DropPartitionResponse;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::convert::to_table_path;
use crate::rpc::frame::{ReadError, WriteError};
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -51,9 +50,7 @@ impl RequestBody for DropPartitionRequest {
type ResponseBody = DropPartitionResponse;
const API_KEY: ApiKey = ApiKey::DropPartition;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(DropPartitionRequest);
-impl_read_version_type!(DropPartitionResponse);
+impl_write_type!(DropPartitionRequest);
+impl_read_type!(DropPartitionResponse);
diff --git a/crates/fluss/src/rpc/message/drop_table.rs
b/crates/fluss/src/rpc/message/drop_table.rs
index a2b3f2d1..b452cf07 100644
--- a/crates/fluss/src/rpc/message/drop_table.rs
+++ b/crates/fluss/src/rpc/message/drop_table.rs
@@ -16,16 +16,15 @@
// under the License.
use crate::metadata::TablePath;
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::{impl_read_type, impl_write_type, proto};
use crate::proto::DropTableResponse;
use crate::rpc::frame::ReadError;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::convert::to_table_path;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -50,9 +49,7 @@ impl RequestBody for DropTableRequest {
type ResponseBody = DropTableResponse;
const API_KEY: ApiKey = ApiKey::DropTable;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(DropTableRequest);
-impl_read_version_type!(DropTableResponse);
+impl_write_type!(DropTableRequest);
+impl_read_type!(DropTableResponse);
diff --git a/crates/fluss/src/rpc/message/fetch.rs
b/crates/fluss/src/rpc/message/fetch.rs
index 15876069..67930f84 100644
--- a/crates/fluss/src/rpc/message/fetch.rs
+++ b/crates/fluss/src/rpc/message/fetch.rs
@@ -19,10 +19,9 @@ use crate::proto::FetchLogResponse;
use crate::rpc::frame::ReadError;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
use prost::Message;
use bytes::{Buf, BufMut};
@@ -50,9 +49,7 @@ impl RequestBody for FetchLogRequest {
type ResponseBody = FetchLogResponse;
const API_KEY: ApiKey = ApiKey::FetchLog;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(FetchLogRequest);
-impl_read_version_type!(FetchLogResponse);
+impl_write_type!(FetchLogRequest);
+impl_read_type!(FetchLogResponse);
diff --git a/crates/fluss/src/rpc/message/get_database_info.rs
b/crates/fluss/src/rpc/message/get_database_info.rs
index 6468bebd..63647d52 100644
--- a/crates/fluss/src/rpc/message/get_database_info.rs
+++ b/crates/fluss/src/rpc/message/get_database_info.rs
@@ -18,10 +18,9 @@
use crate::rpc::frame::ReadError;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -44,8 +43,7 @@ impl RequestBody for GetDatabaseInfoRequest {
type ResponseBody = proto::GetDatabaseInfoResponse;
const API_KEY: ApiKey = ApiKey::GetDatabaseInfo;
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(GetDatabaseInfoRequest);
-impl_read_version_type!(proto::GetDatabaseInfoResponse);
+impl_write_type!(GetDatabaseInfoRequest);
+impl_read_type!(proto::GetDatabaseInfoResponse);
diff --git a/crates/fluss/src/rpc/message/get_latest_lake_snapshot.rs
b/crates/fluss/src/rpc/message/get_latest_lake_snapshot.rs
index a632a159..0b59384d 100644
--- a/crates/fluss/src/rpc/message/get_latest_lake_snapshot.rs
+++ b/crates/fluss/src/rpc/message/get_latest_lake_snapshot.rs
@@ -18,14 +18,13 @@
use crate::proto;
use crate::proto::PbTablePath;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
use crate::metadata::TablePath;
use crate::rpc::frame::ReadError;
-use crate::{impl_read_version_type, impl_write_version_type};
+use crate::{impl_read_type, impl_write_type};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -50,8 +49,7 @@ impl GetLatestLakeSnapshotRequest {
impl RequestBody for GetLatestLakeSnapshotRequest {
type ResponseBody = proto::GetLatestLakeSnapshotResponse;
const API_KEY: ApiKey = ApiKey::GetLatestLakeSnapshot;
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(GetLatestLakeSnapshotRequest);
-impl_read_version_type!(proto::GetLatestLakeSnapshotResponse);
+impl_write_type!(GetLatestLakeSnapshotRequest);
+impl_read_type!(proto::GetLatestLakeSnapshotResponse);
diff --git a/crates/fluss/src/rpc/message/get_security_token.rs
b/crates/fluss/src/rpc/message/get_security_token.rs
index 7995232d..741c8482 100644
--- a/crates/fluss/src/rpc/message/get_security_token.rs
+++ b/crates/fluss/src/rpc/message/get_security_token.rs
@@ -17,10 +17,9 @@
use crate::proto::{GetFileSystemSecurityTokenRequest,
GetFileSystemSecurityTokenResponse};
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::{ReadError, WriteError};
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -46,8 +45,7 @@ impl Default for GetSecurityTokenRequest {
impl RequestBody for GetSecurityTokenRequest {
type ResponseBody = GetFileSystemSecurityTokenResponse;
const API_KEY: ApiKey = ApiKey::GetFileSystemSecurityToken;
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(GetSecurityTokenRequest);
-impl_read_version_type!(GetFileSystemSecurityTokenResponse);
+impl_write_type!(GetSecurityTokenRequest);
+impl_read_type!(GetFileSystemSecurityTokenResponse);
diff --git a/crates/fluss/src/rpc/message/get_table.rs
b/crates/fluss/src/rpc/message/get_table.rs
index 61657f7a..a7562f92 100644
--- a/crates/fluss/src/rpc/message/get_table.rs
+++ b/crates/fluss/src/rpc/message/get_table.rs
@@ -17,14 +17,13 @@
use crate::proto::{GetTableInfoRequest, GetTableInfoResponse, PbTablePath};
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
use crate::metadata::TablePath;
use crate::rpc::frame::ReadError;
-use crate::{impl_read_version_type, impl_write_version_type};
+use crate::{impl_read_type, impl_write_type};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -49,8 +48,7 @@ impl GetTableRequest {
impl RequestBody for GetTableRequest {
type ResponseBody = GetTableInfoResponse;
const API_KEY: ApiKey = ApiKey::GetTable;
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(GetTableRequest);
-impl_read_version_type!(GetTableInfoResponse);
+impl_write_type!(GetTableRequest);
+impl_read_type!(GetTableInfoResponse);
diff --git a/crates/fluss/src/rpc/message/get_table_schema.rs
b/crates/fluss/src/rpc/message/get_table_schema.rs
index ad7b23fb..1c7c00b7 100644
--- a/crates/fluss/src/rpc/message/get_table_schema.rs
+++ b/crates/fluss/src/rpc/message/get_table_schema.rs
@@ -17,14 +17,13 @@
use crate::proto::{GetTableSchemaRequest, GetTableSchemaResponse, PbTablePath};
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
use crate::metadata::TablePath;
use crate::rpc::frame::ReadError;
-use crate::{impl_read_version_type, impl_write_version_type};
+use crate::{impl_read_type, impl_write_type};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -50,8 +49,7 @@ impl GetTableSchemaRequestMsg {
impl RequestBody for GetTableSchemaRequestMsg {
type ResponseBody = GetTableSchemaResponse;
const API_KEY: ApiKey = ApiKey::GetTableSchema;
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(GetTableSchemaRequestMsg);
-impl_read_version_type!(GetTableSchemaResponse);
+impl_write_type!(GetTableSchemaRequestMsg);
+impl_read_type!(GetTableSchemaResponse);
diff --git a/crates/fluss/src/rpc/message/header.rs
b/crates/fluss/src/rpc/message/header.rs
index 2f5848aa..11155f68 100644
--- a/crates/fluss/src/rpc/message/header.rs
+++ b/crates/fluss/src/rpc/message/header.rs
@@ -19,7 +19,7 @@ use crate::proto::ErrorResponse;
use crate::rpc::api_key::ApiKey;
use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::{ReadError, WriteError};
-use crate::rpc::message::{ReadVersionedType, WriteVersionedType};
+use crate::rpc::message::{ReadType, WriteType};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -42,11 +42,11 @@ pub struct RequestHeader {
pub client_id: Option<String>,
}
-impl<W> WriteVersionedType<W> for RequestHeader
+impl<W> WriteType<W> for RequestHeader
where
W: BufMut,
{
- fn write_versioned(&self, writer: &mut W, _version: ApiVersion) ->
Result<(), WriteError> {
+ fn write(&self, writer: &mut W) -> Result<(), WriteError> {
writer.put_i16(self.request_api_key.into());
writer.put_i16(self.request_api_version.0);
writer.put_i32(self.request_id);
@@ -60,11 +60,11 @@ pub struct ResponseHeader {
pub error_response: Option<ErrorResponse>,
}
-impl<R> ReadVersionedType<R> for ResponseHeader
+impl<R> ReadType<R> for ResponseHeader
where
R: Buf,
{
- fn read_versioned(reader: &mut R, _version: ApiVersion) -> Result<Self,
ReadError> {
+ fn read(reader: &mut R) -> Result<Self, ReadError> {
let resp_type = reader.get_u8();
let request_id = reader.get_i32();
if resp_type != SUCCESS_RESPONSE {
diff --git a/crates/fluss/src/rpc/message/init_writer.rs
b/crates/fluss/src/rpc/message/init_writer.rs
index 0bbb0dc5..b2e64a5f 100644
--- a/crates/fluss/src/rpc/message/init_writer.rs
+++ b/crates/fluss/src/rpc/message/init_writer.rs
@@ -17,10 +17,9 @@
use crate::proto::{InitWriterResponse, PbTablePath};
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::{ReadError, WriteError};
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -42,9 +41,7 @@ impl RequestBody for InitWriterRequest {
type ResponseBody = InitWriterResponse;
const API_KEY: ApiKey = ApiKey::InitWriter;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(InitWriterRequest);
-impl_read_version_type!(InitWriterResponse);
+impl_write_type!(InitWriterRequest);
+impl_read_type!(InitWriterResponse);
diff --git a/crates/fluss/src/rpc/message/limit_scan.rs
b/crates/fluss/src/rpc/message/limit_scan.rs
index d83a2e8b..c71b03c3 100644
--- a/crates/fluss/src/rpc/message/limit_scan.rs
+++ b/crates/fluss/src/rpc/message/limit_scan.rs
@@ -19,10 +19,9 @@ use crate::proto::LimitScanResponse;
use crate::rpc::frame::ReadError;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
use prost::Message;
use bytes::{Buf, BufMut};
@@ -50,9 +49,7 @@ impl RequestBody for LimitScanRequest {
type ResponseBody = LimitScanResponse;
const API_KEY: ApiKey = ApiKey::LimitScan;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(LimitScanRequest);
-impl_read_version_type!(LimitScanResponse);
+impl_write_type!(LimitScanRequest);
+impl_read_type!(LimitScanResponse);
diff --git a/crates/fluss/src/rpc/message/list_databases.rs
b/crates/fluss/src/rpc/message/list_databases.rs
index 83226ab1..21e16400 100644
--- a/crates/fluss/src/rpc/message/list_databases.rs
+++ b/crates/fluss/src/rpc/message/list_databases.rs
@@ -18,10 +18,9 @@
use crate::rpc::frame::ReadError;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -42,8 +41,7 @@ impl RequestBody for ListDatabasesRequest {
type ResponseBody = proto::ListDatabasesResponse;
const API_KEY: ApiKey = ApiKey::ListDatabases;
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(ListDatabasesRequest);
-impl_read_version_type!(proto::ListDatabasesResponse);
+impl_write_type!(ListDatabasesRequest);
+impl_read_type!(proto::ListDatabasesResponse);
diff --git a/crates/fluss/src/rpc/message/list_offsets.rs
b/crates/fluss/src/rpc/message/list_offsets.rs
index 262645a6..2ec14370 100644
--- a/crates/fluss/src/rpc/message/list_offsets.rs
+++ b/crates/fluss/src/rpc/message/list_offsets.rs
@@ -15,9 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{
- BucketId, PartitionId, TableId, impl_read_version_type,
impl_write_version_type, proto,
-};
+use crate::{BucketId, PartitionId, TableId, impl_read_type, impl_write_type,
proto};
use crate::error::Result as FlussResult;
use crate::error::{Error, FlussError};
@@ -25,9 +23,8 @@ use crate::proto::{ErrorResponse, ListOffsetsResponse};
use crate::rpc::frame::ReadError;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
use std::collections::HashMap;
use bytes::{Buf, BufMut};
@@ -98,12 +95,10 @@ impl RequestBody for ListOffsetsRequest {
type ResponseBody = ListOffsetsResponse;
const API_KEY: ApiKey = ApiKey::ListOffsets;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(ListOffsetsRequest);
-impl_read_version_type!(ListOffsetsResponse);
+impl_write_type!(ListOffsetsRequest);
+impl_read_type!(ListOffsetsResponse);
impl ListOffsetsResponse {
pub fn offsets(&self) -> FlussResult<HashMap<i32, i64>> {
diff --git a/crates/fluss/src/rpc/message/list_partition_infos.rs
b/crates/fluss/src/rpc/message/list_partition_infos.rs
index ab693671..cf24f466 100644
--- a/crates/fluss/src/rpc/message/list_partition_infos.rs
+++ b/crates/fluss/src/rpc/message/list_partition_infos.rs
@@ -18,11 +18,10 @@
use crate::metadata::{PartitionInfo, PartitionSpec, TablePath};
use crate::proto::ListPartitionInfosResponse;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::convert::to_table_path;
use crate::rpc::frame::{ReadError, WriteError};
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -46,12 +45,10 @@ impl RequestBody for ListPartitionInfosRequest {
type ResponseBody = ListPartitionInfosResponse;
const API_KEY: ApiKey = ApiKey::ListPartitionInfos;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(ListPartitionInfosRequest);
-impl_read_version_type!(ListPartitionInfosResponse);
+impl_write_type!(ListPartitionInfosRequest);
+impl_read_type!(ListPartitionInfosResponse);
impl ListPartitionInfosResponse {
pub fn get_partitions_info(&self) -> Vec<PartitionInfo> {
diff --git a/crates/fluss/src/rpc/message/list_tables.rs
b/crates/fluss/src/rpc/message/list_tables.rs
index ff2497a0..8ff72141 100644
--- a/crates/fluss/src/rpc/message/list_tables.rs
+++ b/crates/fluss/src/rpc/message/list_tables.rs
@@ -15,15 +15,14 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::{impl_read_type, impl_write_type, proto};
use crate::proto::ListTablesResponse;
use crate::rpc::frame::ReadError;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -47,9 +46,7 @@ impl RequestBody for ListTablesRequest {
type ResponseBody = ListTablesResponse;
const API_KEY: ApiKey = ApiKey::ListTables;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(ListTablesRequest);
-impl_read_version_type!(ListTablesResponse);
+impl_write_type!(ListTablesRequest);
+impl_read_type!(ListTablesResponse);
diff --git a/crates/fluss/src/rpc/message/lookup.rs
b/crates/fluss/src/rpc/message/lookup.rs
index 07f8e06c..200d4bc8 100644
--- a/crates/fluss/src/rpc/message/lookup.rs
+++ b/crates/fluss/src/rpc/message/lookup.rs
@@ -19,12 +19,9 @@ use crate::proto::LookupResponse;
use crate::rpc::frame::ReadError;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{
- BucketId, PartitionId, TableId, impl_read_version_type,
impl_write_version_type, proto,
-};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{BucketId, PartitionId, TableId, impl_read_type, impl_write_type,
proto};
use bytes::Bytes;
use prost::Message;
@@ -65,9 +62,7 @@ impl RequestBody for LookupRequest {
type ResponseBody = LookupResponse;
const API_KEY: ApiKey = ApiKey::Lookup;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(LookupRequest);
-impl_read_version_type!(LookupResponse);
+impl_write_type!(LookupRequest);
+impl_read_type!(LookupResponse);
diff --git a/crates/fluss/src/rpc/message/mod.rs
b/crates/fluss/src/rpc/message/mod.rs
index 456f9cb8..096066ed 100644
--- a/crates/fluss/src/rpc/message/mod.rs
+++ b/crates/fluss/src/rpc/message/mod.rs
@@ -16,10 +16,10 @@
// under the License.
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::{ReadError, WriteError};
use bytes::{Buf, BufMut};
+mod api_versions;
mod authenticate;
mod create_database;
mod create_partition;
@@ -49,6 +49,7 @@ mod table_exists;
mod update_metadata;
pub use crate::rpc::RpcError;
+pub use api_versions::*;
pub use authenticate::*;
pub use create_database::*;
pub use create_partition::*;
@@ -81,44 +82,36 @@ pub trait RequestBody {
type ResponseBody;
const API_KEY: ApiKey;
-
- const REQUEST_VERSION: ApiVersion;
}
impl<T: RequestBody> RequestBody for &T {
type ResponseBody = T::ResponseBody;
const API_KEY: ApiKey = T::API_KEY;
-
- const REQUEST_VERSION: ApiVersion = T::REQUEST_VERSION;
}
-pub trait WriteVersionedType<W>: Sized
+pub trait WriteType<W>: Sized
where
W: BufMut,
{
- fn write_versioned(&self, writer: &mut W, version: ApiVersion) ->
Result<(), WriteError>;
+ fn write(&self, writer: &mut W) -> Result<(), WriteError>;
}
-pub trait ReadVersionedType<R>: Sized
+pub trait ReadType<R>: Sized
where
R: Buf,
{
- fn read_versioned(reader: &mut R, version: ApiVersion) -> Result<Self,
ReadError>;
+ fn read(reader: &mut R) -> Result<Self, ReadError>;
}
#[macro_export]
-macro_rules! impl_write_version_type {
+macro_rules! impl_write_type {
($type:ty) => {
- impl<W> WriteVersionedType<W> for $type
+ impl<W> WriteType<W> for $type
where
W: BufMut,
{
- fn write_versioned(
- &self,
- writer: &mut W,
- _version: ApiVersion,
- ) -> Result<(), WriteError> {
+ fn write(&self, writer: &mut W) -> Result<(), WriteError> {
Ok(self.inner_request.encode(writer).unwrap())
}
}
@@ -126,13 +119,13 @@ macro_rules! impl_write_version_type {
}
#[macro_export]
-macro_rules! impl_read_version_type {
+macro_rules! impl_read_type {
($type:ty) => {
- impl<R> ReadVersionedType<R> for $type
+ impl<R> ReadType<R> for $type
where
R: Buf,
{
- fn read_versioned(reader: &mut R, _version: ApiVersion) ->
Result<Self, ReadError> {
+ fn read(reader: &mut R) -> Result<Self, ReadError> {
Ok(<$type>::decode(reader).unwrap())
}
}
diff --git a/crates/fluss/src/rpc/message/prefix_lookup.rs
b/crates/fluss/src/rpc/message/prefix_lookup.rs
index 5ee44d25..e71ffe7c 100644
--- a/crates/fluss/src/rpc/message/prefix_lookup.rs
+++ b/crates/fluss/src/rpc/message/prefix_lookup.rs
@@ -19,12 +19,9 @@ use crate::proto::PrefixLookupResponse;
use crate::rpc::frame::ReadError;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{
- BucketId, PartitionId, TableId, impl_read_version_type,
impl_write_version_type, proto,
-};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{BucketId, PartitionId, TableId, impl_read_type, impl_write_type,
proto};
use bytes::Bytes;
use prost::Message;
@@ -65,9 +62,7 @@ impl RequestBody for PrefixLookupRequest {
type ResponseBody = PrefixLookupResponse;
const API_KEY: ApiKey = ApiKey::PrefixLookup;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(PrefixLookupRequest);
-impl_read_version_type!(PrefixLookupResponse);
+impl_write_type!(PrefixLookupRequest);
+impl_read_type!(PrefixLookupResponse);
diff --git a/crates/fluss/src/rpc/message/produce_log.rs
b/crates/fluss/src/rpc/message/produce_log.rs
index dab7ea9a..8be24638 100644
--- a/crates/fluss/src/rpc/message/produce_log.rs
+++ b/crates/fluss/src/rpc/message/produce_log.rs
@@ -21,10 +21,9 @@ use crate::rpc::frame::ReadError;
use crate::client::ReadyWriteBatch;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -63,9 +62,7 @@ impl RequestBody for ProduceLogRequest {
type ResponseBody = ProduceLogResponse;
const API_KEY: ApiKey = ApiKey::ProduceLog;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(ProduceLogRequest);
-impl_read_version_type!(ProduceLogResponse);
+impl_write_type!(ProduceLogRequest);
+impl_read_type!(ProduceLogResponse);
diff --git a/crates/fluss/src/rpc/message/put_kv.rs
b/crates/fluss/src/rpc/message/put_kv.rs
index 983faa66..e76496d1 100644
--- a/crates/fluss/src/rpc/message/put_kv.rs
+++ b/crates/fluss/src/rpc/message/put_kv.rs
@@ -18,11 +18,10 @@
use crate::client::ReadyWriteBatch;
use crate::proto::{PbPutKvReqForBucket, PutKvResponse};
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::ReadError;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -65,9 +64,7 @@ impl RequestBody for PutKvRequest {
type ResponseBody = PutKvResponse;
const API_KEY: ApiKey = ApiKey::PutKv;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(PutKvRequest);
-impl_read_version_type!(PutKvResponse);
+impl_write_type!(PutKvRequest);
+impl_read_type!(PutKvResponse);
diff --git a/crates/fluss/src/rpc/message/table_exists.rs
b/crates/fluss/src/rpc/message/table_exists.rs
index ec982116..5bc848e3 100644
--- a/crates/fluss/src/rpc/message/table_exists.rs
+++ b/crates/fluss/src/rpc/message/table_exists.rs
@@ -16,14 +16,13 @@
// under the License.
use crate::metadata::TablePath;
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::{impl_read_type, impl_write_type, proto};
use crate::proto::TableExistsResponse;
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::convert::to_table_path;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
use crate::rpc::frame::ReadError;
@@ -48,9 +47,7 @@ impl RequestBody for TableExistsRequest {
type ResponseBody = TableExistsResponse;
const API_KEY: ApiKey = ApiKey::TableExists;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(TableExistsRequest);
-impl_read_version_type!(TableExistsResponse);
+impl_write_type!(TableExistsRequest);
+impl_read_type!(TableExistsResponse);
diff --git a/crates/fluss/src/rpc/message/update_metadata.rs
b/crates/fluss/src/rpc/message/update_metadata.rs
index 1f0d88c2..fd96ca5e 100644
--- a/crates/fluss/src/rpc/message/update_metadata.rs
+++ b/crates/fluss/src/rpc/message/update_metadata.rs
@@ -18,14 +18,13 @@
use crate::metadata::{PhysicalTablePath, TablePath};
use crate::proto::{MetadataResponse, PbPhysicalTablePath, PbTablePath};
use crate::rpc::api_key::ApiKey;
-use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::ReadError;
use crate::rpc::frame::WriteError;
-use crate::rpc::message::{ReadVersionedType, RequestBody, WriteVersionedType};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
use std::collections::HashSet;
use std::sync::Arc;
-use crate::{impl_read_version_type, impl_write_version_type, proto};
+use crate::{impl_read_type, impl_write_type, proto};
use bytes::{Buf, BufMut};
use prost::Message;
@@ -66,9 +65,7 @@ impl RequestBody for UpdateMetadataRequest {
type ResponseBody = MetadataResponse;
const API_KEY: ApiKey = ApiKey::MetaData;
-
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
-impl_write_version_type!(UpdateMetadataRequest);
-impl_read_version_type!(MetadataResponse);
+impl_write_type!(UpdateMetadataRequest);
+impl_read_type!(MetadataResponse);
diff --git a/crates/fluss/src/rpc/server_connection.rs
b/crates/fluss/src/rpc/server_connection.rs
index 46d99c0d..e1148f9d 100644
--- a/crates/fluss/src/rpc/server_connection.rs
+++ b/crates/fluss/src/rpc/server_connection.rs
@@ -15,20 +15,22 @@
// specific language governing permissions and limitations
// under the License.
-use crate::cluster::ServerNode;
+use crate::cluster::{ServerNode, ServerType};
use crate::error::Error;
use crate::metrics::{
CLIENT_BYTES_RECEIVED_TOTAL, CLIENT_BYTES_SENT_TOTAL,
CLIENT_REQUEST_LATENCY_MS,
CLIENT_REQUESTS_IN_FLIGHT, CLIENT_REQUESTS_TOTAL, CLIENT_RESPONSES_TOTAL,
LABEL_API_KEY,
api_key_label,
};
+use crate::proto::PbApiVersion;
+use crate::rpc::api_key::ApiKey;
use crate::rpc::api_version::ApiVersion;
use crate::rpc::error::RpcError;
use crate::rpc::error::RpcError::ConnectionError;
use crate::rpc::frame::{AsyncMessageRead, AsyncMessageWrite};
use crate::rpc::message::{
- REQUEST_HEADER_LENGTH, ReadVersionedType, RequestBody, RequestHeader,
ResponseHeader,
- WriteVersionedType,
+ ApiVersionsRequest, REQUEST_HEADER_LENGTH, ReadType, RequestBody,
RequestHeader,
+ ResponseHeader, WriteType,
};
use crate::rpc::transport::Transport;
use futures::future::BoxFuture;
@@ -72,6 +74,108 @@ impl fmt::Debug for SaslConfig {
}
}
+/// Represents the negotiated API versions between the client and a server
node.
+/// Built from the server's `ApiVersionsResponse` by intersecting each API's
+/// client-supported range with the server-supported range, keeping the highest
+/// usable version.
+#[derive(Clone, Debug)]
+pub struct ServerApiVersions {
+ versions: HashMap<ApiKey, Result<ApiVersion, String>>,
+}
+
+impl ServerApiVersions {
+ /// Build from the server's advertised API version list.
+ pub fn new(server_versions: &[PbApiVersion]) -> Self {
+ let mut versions = HashMap::new();
+ for sv in server_versions {
+ let api_key = ApiKey::from(i16::try_from(sv.api_key).unwrap());
+ // Skip unknown API keys — the client does not support them.
+ let client_range = match api_key.supported_versions() {
+ Some(range) => range,
+ None => continue,
+ };
+ let server_min = i16::try_from(sv.min_version).unwrap();
+ let server_max = i16::try_from(sv.max_version).unwrap();
+ let min_version = client_range.min().0.max(server_min);
+ let max_version = client_range.max().0.min(server_max);
+ if min_version > max_version {
+ versions.insert(
+ api_key,
+ Err(format!(
+ "The server does not support {:?} with version in
range [{},{}]. \
+ The supported range is [{},{}].",
+ api_key,
+ client_range.min(),
+ client_range.max(),
+ server_min,
+ server_max,
+ )),
+ );
+ } else {
+ versions.insert(api_key, Ok(ApiVersion(max_version)));
+ }
+ }
+ Self { versions }
+ }
+
+ /// Get the negotiated (highest usable) version for a given API key.
+ pub fn highest_available_version(&self, api_key: ApiKey) ->
Result<ApiVersion, Error> {
+ match self.versions.get(&api_key) {
+ Some(Ok(version)) => Ok(*version),
+ Some(Err(msg)) => Err(Error::UnsupportedVersion {
+ message: msg.clone(),
+ }),
+ None => Err(Error::UnsupportedVersion {
+ message: format!("The server does not support {:?}", api_key),
+ }),
+ }
+ }
+}
+
+/// Resolve the API version to use for a given API key.
+fn resolve_api_version_for(
+ api_versions: Option<&ServerApiVersions>,
+ api_key: ApiKey,
+) -> Result<ApiVersion, Error> {
+ // version equals highestSupportedVersion might happen when requesting api
version check
+ // before serverApiVersions is initialized. We always use the highest
version for api
+ // version checking.
+ let default_version = api_key
+ .supported_versions()
+ .map(|range| range.max())
+ .unwrap();
+ match api_versions {
+ Some(versions) => versions.highest_available_version(api_key),
+ None => Ok(default_version),
+ }
+}
+
+/// Validate that the server's advertised `server_type` matches the type we
expect
+/// for the target `ServerNode`.
+fn validate_server_type(
+ expected: &ServerType,
+ response_server_type: Option<i32>,
+) -> Result<(), Error> {
+ // For forward-compat with servers that do not populate `server_type`,
validation is skipped.
+ let Some(type_id) = response_server_type else {
+ return Ok(());
+ };
+ let actual = ServerType::from_type_id(type_id);
+ if actual.as_ref() == Some(expected) {
+ return Ok(());
+ }
+ let actual_desc = actual
+ .map(|t| t.to_string())
+ .unwrap_or_else(|| format!("Unknown(type_id={type_id})"));
+ Err(Error::InvalidServerType {
+ message: format!(
+ "Expected server type {expected} but the server advertised
{actual_desc}. \
+ The client may be talking to the wrong endpoint \
+ (e.g. coordinator vs tablet server)."
+ ),
+ })
+}
+
#[derive(Debug, Default)]
pub struct RpcClient {
connections: RwLock<HashMap<String, ServerConnection>>,
@@ -142,6 +246,9 @@ impl RpcClient {
);
let connection = ServerConnection::new(messenger);
+ // Negotiate API versions (must happen before authentication).
+ Self::check_api_versions(&connection,
server_node.server_type()).await?;
+
if let Some(ref sasl) = self.sasl_config {
Self::authenticate(&connection, &sasl.username,
&sasl.password).await?;
}
@@ -149,6 +256,20 @@ impl RpcClient {
Ok(connection)
}
+ /// Send an `ApiVersionsRequest`, validate the advertised `server_type`,
and
+ /// store the negotiated versions on the connection.
+ async fn check_api_versions(
+ connection: &ServerConnection,
+ expected_server_type: &ServerType,
+ ) -> Result<(), Error> {
+ let request = ApiVersionsRequest::new("fluss-rust",
env!("CARGO_PKG_VERSION"));
+ let response = connection.request(request).await?;
+ validate_server_type(expected_server_type, response.server_type)?;
+ let api_versions = ServerApiVersions::new(&response.api_versions);
+ *connection.api_versions.lock() = Some(api_versions);
+ Ok(())
+ }
+
/// Perform SASL/PLAIN authentication handshake.
///
/// Retries on `RetriableAuthenticateException` with exponential backoff
@@ -325,6 +446,10 @@ pub struct ServerConnectionInner<RW> {
state: Arc<Mutex<ConnectionState>>,
+ /// Negotiated API versions for this connection.
+ /// `None` until the ApiVersions handshake completes.
+ api_versions: Mutex<Option<ServerApiVersions>>,
+
join_handle: JoinHandle<()>,
}
@@ -344,16 +469,13 @@ where
Ok(msg) => {
// message was read, so all subsequent errors should
not poison the whole stream
let mut cursor = Cursor::new(msg);
- let header =
- match ResponseHeader::read_versioned(&mut cursor,
ApiVersion(0)) {
- Ok(header) => header,
- Err(err) => {
- log::warn!(
- "Cannot read message header, ignoring
message: {err:?}"
- );
- continue;
- }
- };
+ let header = match ResponseHeader::read(&mut cursor) {
+ Ok(header) => header,
+ Err(err) => {
+ log::warn!("Cannot read message header,
ignoring message: {err:?}");
+ continue;
+ }
+ };
let active_request = match
state_captured.lock().deref_mut() {
ConnectionState::RequestMap(map) => {
@@ -396,10 +518,16 @@ where
client_id,
request_id: AtomicI32::new(0),
state,
+ api_versions: Mutex::new(None),
join_handle,
}
}
+ fn resolve_api_version(&self, api_key: ApiKey) -> Result<ApiVersion,
Error> {
+ let guard = self.api_versions.lock();
+ resolve_api_version_for(guard.as_ref(), api_key)
+ }
+
fn is_poisoned(&self) -> bool {
let guard = self.state.lock();
matches!(*guard, ConnectionState::Poison(_))
@@ -407,29 +535,25 @@ where
pub async fn request<R>(&self, msg: R) -> Result<R::ResponseBody, Error>
where
- R: RequestBody + Send + WriteVersionedType<Vec<u8>>,
- R::ResponseBody: ReadVersionedType<Cursor<Vec<u8>>>,
+ R: RequestBody + Send + WriteType<Vec<u8>>,
+ R::ResponseBody: ReadType<Cursor<Vec<u8>>>,
{
+ let api_version = self.resolve_api_version(R::API_KEY)?;
let request_id = self.request_id.fetch_add(1, Ordering::SeqCst) &
0x7FFFFFFF;
let header = RequestHeader {
request_api_key: R::API_KEY,
- request_api_version: ApiVersion(0),
+ request_api_version: api_version,
request_id,
client_id: Some(String::from(self.client_id.as_ref())),
};
- let header_version = ApiVersion(0);
-
- let body_api_version = ApiVersion(0);
-
let mut buf = Vec::new();
// write header
header
- .write_versioned(&mut buf, header_version)
+ .write(&mut buf)
.map_err(RpcError::WriteMessageError)?;
// write message body
- msg.write_versioned(&mut buf, body_api_version)
- .map_err(RpcError::WriteMessageError)?;
+ msg.write(&mut buf).map_err(RpcError::WriteMessageError)?;
let (tx, rx) = channel();
@@ -473,8 +597,7 @@ where
});
}
- let body = R::ResponseBody::read_versioned(&mut response.data,
body_api_version)
- .map_err(RpcError::ReadMessageError)?;
+ let body = R::ResponseBody::read(&mut
response.data).map_err(RpcError::ReadMessageError)?;
let read_bytes = response.data.position();
let message_bytes = response.data.into_inner().len() as u64;
@@ -483,7 +606,7 @@ where
message_size: message_bytes,
read: read_bytes,
api_key: R::API_KEY,
- api_version: body_api_version,
+ api_version,
}
.into());
}
@@ -643,7 +766,7 @@ mod tests {
use crate::rpc::ApiKey;
use crate::rpc::api_version::ApiVersion;
use crate::rpc::frame::{ReadError, WriteError};
- use crate::rpc::message::{ReadVersionedType, RequestBody,
WriteVersionedType};
+ use crate::rpc::message::{ReadType, RequestBody, WriteType};
use metrics::{SharedString, Unit};
use metrics_util::CompositeKey;
use metrics_util::debugging::{DebugValue, DebuggingRecorder};
@@ -659,17 +782,16 @@ mod tests {
impl RequestBody for TestProduceRequest {
type ResponseBody = TestProduceResponse;
const API_KEY: ApiKey = ApiKey::ProduceLog;
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
- impl WriteVersionedType<Vec<u8>> for TestProduceRequest {
- fn write_versioned(&self, _w: &mut Vec<u8>, _v: ApiVersion) ->
Result<(), WriteError> {
+ impl WriteType<Vec<u8>> for TestProduceRequest {
+ fn write(&self, _w: &mut Vec<u8>) -> Result<(), WriteError> {
Ok(())
}
}
- impl ReadVersionedType<Cursor<Vec<u8>>> for TestProduceResponse {
- fn read_versioned(_r: &mut Cursor<Vec<u8>>, _v: ApiVersion) ->
Result<Self, ReadError> {
+ impl ReadType<Cursor<Vec<u8>>> for TestProduceResponse {
+ fn read(_r: &mut Cursor<Vec<u8>>) -> Result<Self, ReadError> {
Ok(TestProduceResponse)
}
}
@@ -680,17 +802,16 @@ mod tests {
impl RequestBody for TestMetadataRequest {
type ResponseBody = TestMetadataResponse;
const API_KEY: ApiKey = ApiKey::MetaData;
- const REQUEST_VERSION: ApiVersion = ApiVersion(0);
}
- impl WriteVersionedType<Vec<u8>> for TestMetadataRequest {
- fn write_versioned(&self, _w: &mut Vec<u8>, _v: ApiVersion) ->
Result<(), WriteError> {
+ impl WriteType<Vec<u8>> for TestMetadataRequest {
+ fn write(&self, _w: &mut Vec<u8>) -> Result<(), WriteError> {
Ok(())
}
}
- impl ReadVersionedType<Cursor<Vec<u8>>> for TestMetadataResponse {
- fn read_versioned(_r: &mut Cursor<Vec<u8>>, _v: ApiVersion) ->
Result<Self, ReadError> {
+ impl ReadType<Cursor<Vec<u8>>> for TestMetadataResponse {
+ fn read(_r: &mut Cursor<Vec<u8>>) -> Result<Self, ReadError> {
Ok(TestMetadataResponse)
}
}
@@ -873,6 +994,11 @@ mod tests {
tokio::spawn(mock_echo_server(server));
let conn = ServerConnectionInner::new(BufStream::new(client),
usize::MAX, Arc::from("t"));
+ *conn.api_versions.lock() = Some(ServerApiVersions::new(&[PbApiVersion
{
+ api_key: 1014,
+ min_version: 0,
+ max_version: 0,
+ }]));
let before: Vec<_> = snapshotter.snapshot().into_vec();
let request_before = counter_for_label(&before, CLIENT_REQUESTS_TOTAL,
"produce_log");
@@ -913,6 +1039,11 @@ mod tests {
tokio::spawn(mock_echo_server(server));
let conn = ServerConnectionInner::new(BufStream::new(client),
usize::MAX, Arc::from("t"));
+ *conn.api_versions.lock() = Some(ServerApiVersions::new(&[PbApiVersion
{
+ api_key: 1012,
+ min_version: 0,
+ max_version: 0,
+ }]));
let before: Vec<_> = snapshotter.snapshot().into_vec();
let request_sum_before = counter_sum(&before, CLIENT_REQUESTS_TOTAL);
let response_sum_before = counter_sum(&before, CLIENT_RESPONSES_TOTAL);
@@ -949,6 +1080,11 @@ mod tests {
let (client, server) = tokio::io::duplex(64);
drop(server); // force write failure on request path
let conn = ServerConnectionInner::new(BufStream::new(client),
usize::MAX, Arc::from("t"));
+ *conn.api_versions.lock() = Some(ServerApiVersions::new(&[PbApiVersion
{
+ api_key: 1014,
+ min_version: 0,
+ max_version: 0,
+ }]));
let before: Vec<_> = snapshotter.snapshot().into_vec();
let request_before = counter_for_label(&before, CLIENT_REQUESTS_TOTAL,
"produce_log");
@@ -997,6 +1133,11 @@ mod tests {
tokio::spawn(mock_error_server(server));
let conn = ServerConnectionInner::new(BufStream::new(client),
usize::MAX, Arc::from("t"));
+ *conn.api_versions.lock() = Some(ServerApiVersions::new(&[PbApiVersion
{
+ api_key: 1014,
+ min_version: 0,
+ max_version: 0,
+ }]));
let before: Vec<_> = snapshotter.snapshot().into_vec();
let response_before = counter_for_label(&before,
CLIENT_RESPONSES_TOTAL, "produce_log");
@@ -1030,4 +1171,130 @@ mod tests {
"in-flight gauge must return to zero after API error"
);
}
+
+ #[tokio::test]
+ async fn server_api_versions_negotiation() {
+ assert_eq!(
+ resolve_api_version_for(None, ApiKey::ApiVersion).unwrap(),
+ ApiVersion(0)
+ );
+
+ assert_eq!(
+ resolve_api_version_for(None, ApiKey::PutKv).unwrap(),
+ ApiVersion(0)
+ );
+
+ let server_versions = vec![
+ // PutKv: server v0..v3, client v0 only (v1 key encoding not yet
implemented) → negotiated v0
+ PbApiVersion {
+ api_key: 1016,
+ min_version: 0,
+ max_version: 3,
+ },
+ // ProduceLog: server v0..v2, client v0 only → negotiated v0
+ PbApiVersion {
+ api_key: 1014,
+ min_version: 0,
+ max_version: 2,
+ },
+ // Disjoint: server v5..v7, client v0 only → error
+ PbApiVersion {
+ api_key: 1015,
+ min_version: 5,
+ max_version: 7,
+ },
+ // Unknown key (9999) → skipped
+ PbApiVersion {
+ api_key: 9999,
+ min_version: 0,
+ max_version: 5,
+ },
+ ];
+ let negotiated = ServerApiVersions::new(&server_versions);
+
+ // Successful negotiation cases
+ assert_eq!(
+ negotiated.highest_available_version(ApiKey::PutKv).unwrap(),
+ ApiVersion(0)
+ );
+ assert_eq!(
+ negotiated
+ .highest_available_version(ApiKey::ProduceLog)
+ .unwrap(),
+ ApiVersion(0)
+ );
+
+ // Disjoint range → error
+ assert!(
+ negotiated
+ .highest_available_version(ApiKey::FetchLog)
+ .unwrap_err()
+ .to_string()
+ .contains(&format!(
+ "The server does not support {:?}",
+ ApiKey::FetchLog
+ ))
+ );
+
+ // Unknown key is skipped → not in map → error
+ assert!(
+ negotiated
+ .highest_available_version(ApiKey::Unknown(9999))
+ .is_err()
+ );
+
+ // Key not advertised by server → error
+ assert!(
+ ServerApiVersions::new(&[])
+ .highest_available_version(ApiKey::FetchLog)
+ .is_err()
+ );
+ }
+
+ #[test]
+ fn server_type_validation() {
+ // Happy path: server advertises the expected type.
+ assert!(
+ validate_server_type(
+ &ServerType::CoordinatorServer,
+ Some(ServerType::CoordinatorServer.to_type_id()),
+ )
+ .is_ok()
+ );
+ assert!(
+ validate_server_type(
+ &ServerType::TabletServer,
+ Some(ServerType::TabletServer.to_type_id()),
+ )
+ .is_ok()
+ );
+
+ // Mismatch: connected to a coordinator while expecting a tablet server
+ // (and vice versa).
+ let err = validate_server_type(
+ &ServerType::TabletServer,
+ Some(ServerType::CoordinatorServer.to_type_id()),
+ )
+ .unwrap_err();
+ assert!(
+ matches!(err, Error::InvalidServerType { .. }),
+ "expected InvalidServerType, got: {err:?}"
+ );
+
+ assert!(matches!(
+ validate_server_type(
+ &ServerType::CoordinatorServer,
+ Some(ServerType::TabletServer.to_type_id()),
+ ),
+ Err(Error::InvalidServerType { .. })
+ ));
+
+ validate_server_type(&ServerType::TabletServer, None).ok();
+ // Unknown / unmapped type id still fails, with the raw id surfaced so
+ // operators can diagnose protocol drift.
+ assert!(matches!(
+ validate_server_type(&ServerType::CoordinatorServer, Some(99),),
+ Err(Error::InvalidServerType { .. })
+ ));
+ }
}
diff --git a/crates/fluss/tests/integration/kv_table.rs
b/crates/fluss/tests/integration/kv_table.rs
index 62e206b6..1979b973 100644
--- a/crates/fluss/tests/integration/kv_table.rs
+++ b/crates/fluss/tests/integration/kv_table.rs
@@ -1404,4 +1404,75 @@ mod kv_table_test {
.await
.expect("Failed to drop table");
}
+
+ /// Test that KV format v2 tables with non-default bucket key reject v0
clients.
+ /// The Rust client currently only supports API version 0 for
PutKv/Lookup/PrefixLookup.
+ /// When the server creates a table with kv_format_version=2 and a
non-default bucket key,
+ /// it rejects v0 clients because CompactedKeyEncoder (v1) is required.
+ // TODO(key-encoding-v1): Once v1 key encoding is implemented and the
client advertises
+ // PutKv/Lookup/PrefixLookup v1, this test should be updated to verify
that v1 clients
+ // can successfully write to and read from kv_format_v2 tables with
non-default bucket keys.
+ #[tokio::test]
+ async fn kv_format_v2_table_rejects_v0_client() {
+ let cluster = get_shared_cluster();
+ let connection = cluster.get_fluss_connection().await;
+
+ let admin = connection.get_admin().unwrap();
+
+ let table_path = TablePath::new("fluss",
"test_kv_format_v2_reject_v0");
+
+ // Create a KV table with:
+ // 1. kv_format_version = 2
+ // 2. non-default bucket key ("a" is a subset of pk ("a", "b"))
+ // 3. datalake format is exist.
+ let table_descriptor = TableDescriptor::builder()
+ .schema(
+ Schema::builder()
+ .column("a", DataTypes::int())
+ .column("b", DataTypes::string())
+ .column("c", DataTypes::string())
+ .primary_key(vec!["a", "b"])
+ .build()
+ .expect("Failed to build schema"),
+ )
+ .distributed_by(Some(2), vec!["a".to_string()])
+ .property("table.kv.format-version", "2")
+ .property("table.datalake.format", "lance")
+ .build()
+ .expect("Failed to build table");
+
+ create_table(&admin, &table_path, &table_descriptor).await;
+
+ let table = connection.get_table(&table_path).await.unwrap();
+
+ // Test PutKv with v0 client - should fail with UNSUPPORTED_VERSION
+ let table_upsert = table.new_upsert().expect("Failed to create
upsert");
+ let upsert_writer = table_upsert
+ .create_writer()
+ .expect("Failed to create writer");
+
+ let mut row = GenericRow::new(3);
+ row.set_field(0, 1);
+ row.set_field(1, "a");
+ row.set_field(2, "value1");
+ let upsert_result = upsert_writer
+ .upsert(&row)
+ .expect("Failed to upsert row")
+ .await;
+ assert!(
+ upsert_result.is_err(),
+ "PutKv with v0 client should be rejected for kv_format_v2 table
with non-default bucket key"
+ );
+ let err_msg = upsert_result.unwrap_err().to_string();
+ assert!(
+ err_msg.contains("Client API version 0 is not supported"),
+ "Expected 'Client API version 0 is not supported' error, got: {}",
+ err_msg
+ );
+
+ admin
+ .drop_table(&table_path, false)
+ .await
+ .expect("Failed to drop table");
+ }
}
