This is an automated email from the ASF dual-hosted git repository.
leekeiabstraction 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 a3a3a50 fix: [bindings] map RpcError to NetworkException for Java
parity (#519)
a3a3a50 is described below
commit a3a3a50b15b8b63824301c02718d32f95b70e2a8
Author: Anton Borisov <[email protected]>
AuthorDate: Sun May 3 12:58:55 2026 +0100
fix: [bindings] map RpcError to NetworkException for Java parity (#519)
---
bindings/cpp/src/lib.rs | 13 ++++++++-----
bindings/elixir/native/fluss_nif/src/atoms.rs | 5 +++++
bindings/elixir/test/error_test.exs | 6 +++---
bindings/python/src/error.rs | 12 ++++++------
4 files changed, 22 insertions(+), 14 deletions(-)
diff --git a/bindings/cpp/src/lib.rs b/bindings/cpp/src/lib.rs
index facb0e3..875373b 100644
--- a/bindings/cpp/src/lib.rs
+++ b/bindings/cpp/src/lib.rs
@@ -23,6 +23,8 @@ use std::time::Duration;
use fluss as fcore;
use fluss::PartitionId;
+use fluss::error::Error;
+use fluss::rpc::FlussError as CoreFlussError;
static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
tokio::runtime::Builder::new_multi_thread()
@@ -855,13 +857,14 @@ fn client_err(msg: String) -> ffi::FfiResult {
err_result(CLIENT_ERROR_CODE, msg)
}
-/// Convert a core Error to FfiResult.
-/// `FlussAPIError` variants carry the server protocol error code directly.
-/// All other error kinds are client-side and use CLIENT_ERROR_CODE.
-fn err_from_core_error(e: &fcore::error::Error) -> ffi::FfiResult {
- use fcore::error::Error;
+fn err_from_core_error(e: &Error) -> ffi::FfiResult {
+ // Transport failures map to `NetworkException` (Java parity,
+ // retriable).
match e {
Error::FlussAPIError { api_error } => err_result(api_error.code,
api_error.message.clone()),
+ Error::RpcError { .. } => {
+ err_result(CoreFlussError::NetworkException.code(), e.to_string())
+ }
_ => client_err(e.to_string()),
}
}
diff --git a/bindings/elixir/native/fluss_nif/src/atoms.rs
b/bindings/elixir/native/fluss_nif/src/atoms.rs
index 0a8e95b..45d5aa3 100644
--- a/bindings/elixir/native/fluss_nif/src/atoms.rs
+++ b/bindings/elixir/native/fluss_nif/src/atoms.rs
@@ -120,10 +120,15 @@ pub struct NifFlussError {
impl NifFlussError {
pub fn from_core(error: &CoreError) -> Self {
+ // Transport failures map to `:network_exception` (Java parity,
+ // retriable).
let (code, error_code) = match error {
CoreError::FlussAPIError { api_error } => {
(api_error_atom(api_error.code), api_error.code)
}
+ CoreError::RpcError { .. } => {
+ (network_exception(), FlussError::NetworkException.code())
+ }
_ => (client_error(), CLIENT_ERROR_CODE),
};
Self {
diff --git a/bindings/elixir/test/error_test.exs
b/bindings/elixir/test/error_test.exs
index 9294391..d6d4017 100644
--- a/bindings/elixir/test/error_test.exs
+++ b/bindings/elixir/test/error_test.exs
@@ -69,17 +69,17 @@ defmodule Fluss.ErrorTest do
end
describe "NIF error surface" do
- test "unreachable server returns %Fluss.Error{code: :client_error,
error_code: -2}" do
+ test "unreachable server returns %Fluss.Error{code: :network_exception,
error_code: 1}" do
config = Fluss.Config.new("127.0.0.1:1")
- assert {:error, %Fluss.Error{code: :client_error, error_code: -2}} =
+ assert {:error, %Fluss.Error{code: :network_exception, error_code: 1}} =
Fluss.Connection.new(config)
end
test "bang variant raises %Fluss.Error{}" do
config = Fluss.Config.new("127.0.0.1:1")
- assert_raise Fluss.Error, ~r/\[client_error\]/, fn ->
+ assert_raise Fluss.Error, ~r/\[network_exception\]/, fn ->
Fluss.Connection.new!(config)
end
end
diff --git a/bindings/python/src/error.rs b/bindings/python/src/error.rs
index 10c6cfa..9d718aa 100644
--- a/bindings/python/src/error.rs
+++ b/bindings/python/src/error.rs
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+use fluss::error::Error;
+use fluss::rpc::FlussError as CoreFlussError;
use pyo3::exceptions::PyException;
use pyo3::prelude::*;
@@ -55,7 +57,6 @@ impl FlussError {
/// Returns ``True`` if retrying the request may succeed. Client-side
errors always return ``False``.
#[getter]
fn is_retriable(&self) -> bool {
- use fluss::rpc::FlussError as CoreFlussError;
if self.error_code == CLIENT_ERROR_CODE {
return false;
}
@@ -68,13 +69,12 @@ impl FlussError {
PyErr::new::<FlussError, _>((message.to_string(), CLIENT_ERROR_CODE))
}
- /// Create a PyErr from a core Error.
- /// `FlussAPIError` variants carry the server protocol error code directly.
- /// All other error kinds are client-side and use CLIENT_ERROR_CODE.
- pub fn from_core_error(error: &fluss::error::Error) -> PyErr {
- use fluss::error::Error;
+ pub fn from_core_error(error: &Error) -> PyErr {
+ // Transport failures map to `NetworkException` (Java parity,
+ // retriable).
let (msg, code) = match error {
Error::FlussAPIError { api_error } => (api_error.message.clone(),
api_error.code),
+ Error::RpcError { .. } => (error.to_string(),
CoreFlussError::NetworkException.code()),
_ => (error.to_string(), CLIENT_ERROR_CODE),
};
PyErr::new::<FlussError, _>((msg, code))
