You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "loserwang1024 (via GitHub)" <gi...@apache.org> on 2026/05/06 13:04:55 UTC
[PR] [rust] Support API_VERSION in rust client. [fluss-rust]
loserwang1024 opened a new pull request, #531:
URL: https://github.com/apache/fluss-rust/pull/531
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss-rust/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
**(The sections below can be removed for hotfixes or typos)**
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #516
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Support API_VERSION in rust client. [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni merged PR #531:
URL: https://github.com/apache/fluss-rust/pull/531
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Support API_VERSION in rust client. [fluss-rust]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #531:
URL: https://github.com/apache/fluss-rust/pull/531#discussion_r3212776675
##########
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;
Review Comment:
add the check and test
##########
crates/fluss/src/rpc/server_connection.rs:
##########
@@ -72,6 +74,89 @@ 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.
+///
+/// The `ApiVersions` request itself always uses `ApiVersion(0)` since it is
+/// sent before the version negotiation handshake.
+/// For all other requests, the handshake must have been completed; otherwise
+/// an error is returned to prevent silent fallback to an incorrect version.
+fn resolve_api_version_for(
+ api_versions: Option<&ServerApiVersions>,
+ api_key: ApiKey,
+) -> Result<ApiVersion, Error> {
+ if api_key == ApiKey::ApiVersion {
+ return Ok(ApiVersion(0));
Review Comment:
done it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Support API_VERSION in rust client. [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #531:
URL: https://github.com/apache/fluss-rust/pull/531#discussion_r3212867407
##########
crates/fluss/src/rpc/server_connection.rs:
##########
@@ -72,6 +74,89 @@ 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());
Review Comment:
i filed an issue: https://github.com/apache/fluss-rust/issues/535
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Support API_VERSION in rust client. [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #531:
URL: https://github.com/apache/fluss-rust/pull/531#discussion_r3212823949
##########
crates/fluss/src/rpc/server_connection.rs:
##########
@@ -72,6 +74,89 @@ 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());
Review Comment:
somehow it got fixed in f7761c4 and then reappeared :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Support API_VERSION in rust client. [fluss-rust]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #531:
URL: https://github.com/apache/fluss-rust/pull/531#discussion_r3212855997
##########
crates/fluss/src/rpc/server_connection.rs:
##########
@@ -1030,4 +1171,126 @@ 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 { .. })
+ ));
+
+ // Unknown / unmapped type id still fails, with the raw id surfaced so
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Support API_VERSION in rust client. [fluss-rust]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #531:
URL: https://github.com/apache/fluss-rust/pull/531#issuecomment-4395493843
@luoyuxia @fresh-borzoni , CC
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Support API_VERSION in rust client. [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #531:
URL: https://github.com/apache/fluss-rust/pull/531#issuecomment-4412087116
@loserwang1024 Ty, merging
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Support API_VERSION in rust client. [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #531:
URL: https://github.com/apache/fluss-rust/pull/531#discussion_r3208664301
##########
crates/fluss/src/rpc/server_connection.rs:
##########
@@ -72,6 +74,89 @@ 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());
Review Comment:
it can potentially panic, Java truncates to short, shall we do smth similar?
##########
crates/fluss/src/rpc/server_connection.rs:
##########
@@ -72,6 +74,89 @@ 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.
+///
+/// The `ApiVersions` request itself always uses `ApiVersion(0)` since it is
+/// sent before the version negotiation handshake.
+/// For all other requests, the handshake must have been completed; otherwise
+/// an error is returned to prevent silent fallback to an incorrect version.
+fn resolve_api_version_for(
+ api_versions: Option<&ServerApiVersions>,
+ api_key: ApiKey,
+) -> Result<ApiVersion, Error> {
+ if api_key == ApiKey::ApiVersion {
+ return Ok(ApiVersion(0));
Review Comment:
I think Java returns apiKey.highestSupportedVersion, it will desync eventually
##########
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;
Review Comment:
`server_type` is added but never read. Java validates it and disconnects
on coordinator/tablet mismatch. Let's file a follow-up, just so we don't forget about it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Support API_VERSION in rust client. [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #531:
URL: https://github.com/apache/fluss-rust/pull/531#discussion_r3212827427
##########
crates/fluss/src/rpc/server_connection.rs:
##########
@@ -1030,4 +1171,126 @@ 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 { .. })
+ ));
+
+ // Unknown / unmapped type id still fails, with the raw id surfaced so
Review Comment:
None is the skip case, not the unknown case. Probably we want two assertions: one for None (returns Ok) and one for Some(99) (returns Err)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org