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 7d71d6ba [elixir] feat: Extend Fluss.Config with security, SASL and
connect timeout options (#577)
7d71d6ba is described below
commit 7d71d6ba3f59e96da7d39ec7eab3227295a45af5
Author: Nicoleta Lazar <[email protected]>
AuthorDate: Fri May 29 00:36:13 2026 +0100
[elixir] feat: Extend Fluss.Config with security, SASL and connect timeout
options (#577)
* feat(elixir): Extend NifConfig with security & connect timeout fields
This commit adds 5 fields to NifConfig: connect timeout ms and
security_* settings (protocol, sasl mechanism, usernama and password)
* feat(elixir): Expose security opts and connect timeout in Fluss.Config
This commit adds 5 fields to Fluss.Config with the corresponding
setters: connect timeout, security protocol, sasl mechanism, username
and password.
To ensure that the password is redacted, we add a custom Inspect impl,
essentially mirroring the upstream rust debug redaction.
---
bindings/elixir/lib/fluss/config.ex | 54 +++++++++++++++++++++++++
bindings/elixir/native/fluss_nif/src/config.rs | 20 +++++++++
bindings/elixir/test/config_test.exs | 56 ++++++++++++++++++++++++++
3 files changed, 130 insertions(+)
diff --git a/bindings/elixir/lib/fluss/config.ex
b/bindings/elixir/lib/fluss/config.ex
index 07324df0..8aaacf79 100644
--- a/bindings/elixir/lib/fluss/config.ex
+++ b/bindings/elixir/lib/fluss/config.ex
@@ -33,6 +33,7 @@ defmodule Fluss.Config do
@enforce_keys [:bootstrap_servers]
defstruct bootstrap_servers: nil,
+ connect_timeout_ms: nil,
remote_file_download_thread_num: nil,
scanner_log_fetch_max_bytes: nil,
scanner_log_fetch_max_bytes_for_bucket: nil,
@@ -41,6 +42,10 @@ defmodule Fluss.Config do
scanner_log_max_poll_records: nil,
scanner_remote_log_prefetch_num: nil,
scanner_remote_log_read_concurrency: nil,
+ security_protocol: nil,
+ security_sasl_mechanism: nil,
+ security_sasl_password: nil,
+ security_sasl_username: nil,
writer_acks: nil,
writer_batch_size: nil,
writer_batch_timeout_ms: nil,
@@ -56,6 +61,7 @@ defmodule Fluss.Config do
@type t :: %__MODULE__{
bootstrap_servers: String.t(),
+ connect_timeout_ms: non_neg_integer() | nil,
remote_file_download_thread_num: non_neg_integer() | nil,
scanner_log_fetch_max_bytes: non_neg_integer() | nil,
scanner_log_fetch_max_bytes_for_bucket: non_neg_integer() | nil,
@@ -64,6 +70,10 @@ defmodule Fluss.Config do
scanner_log_max_poll_records: non_neg_integer() | nil,
scanner_remote_log_prefetch_num: non_neg_integer() | nil,
scanner_remote_log_read_concurrency: non_neg_integer() | nil,
+ security_protocol: String.t() | nil,
+ security_sasl_mechanism: String.t() | nil,
+ security_sasl_password: String.t() | nil,
+ security_sasl_username: String.t() | nil,
writer_acks: String.t() | nil,
writer_batch_size: non_neg_integer() | nil,
writer_batch_timeout_ms: non_neg_integer() | nil,
@@ -90,6 +100,10 @@ defmodule Fluss.Config do
def set_bootstrap_servers(%__MODULE__{} = config, servers) when
is_binary(servers),
do: %{config | bootstrap_servers: servers}
+ @spec set_connect_timeout_ms(t(), non_neg_integer()) :: t()
+ def set_connect_timeout_ms(%__MODULE__{} = config, ms) when is_integer(ms),
+ do: %{config | connect_timeout_ms: ms}
+
@spec set_remote_file_download_thread_num(t(), non_neg_integer()) :: t()
def set_remote_file_download_thread_num(%__MODULE__{} = config, threads)
when is_integer(threads),
@@ -128,6 +142,22 @@ defmodule Fluss.Config do
when is_integer(concurrency),
do: %{config | scanner_remote_log_read_concurrency: concurrency}
+ @spec set_security_protocol(t(), String.t()) :: t()
+ def set_security_protocol(%__MODULE__{} = config, protocol) when
is_binary(protocol),
+ do: %{config | security_protocol: protocol}
+
+ @spec set_security_sasl_mechanism(t(), String.t()) :: t()
+ def set_security_sasl_mechanism(%__MODULE__{} = config, mechanism) when
is_binary(mechanism),
+ do: %{config | security_sasl_mechanism: mechanism}
+
+ @spec set_security_sasl_password(t(), String.t()) :: t()
+ def set_security_sasl_password(%__MODULE__{} = config, pass) when
is_binary(pass),
+ do: %{config | security_sasl_password: pass}
+
+ @spec set_security_sasl_username(t(), String.t()) :: t()
+ def set_security_sasl_username(%__MODULE__{} = config, username) when
is_binary(username),
+ do: %{config | security_sasl_username: username}
+
@spec set_writer_acks(t(), String.t()) :: t()
def set_writer_acks(%__MODULE__{} = config, acks) when is_binary(acks),
do: %{config | writer_acks: acks}
@@ -183,3 +213,27 @@ defmodule Fluss.Config do
@spec get_bootstrap_servers(t()) :: String.t()
def get_bootstrap_servers(%__MODULE__{bootstrap_servers: servers}), do:
servers
end
+
+defimpl Inspect, for: Fluss.Config do
+ import Inspect.Algebra
+
+ def inspect(%Fluss.Config{} = config, opts) do
+ sanitized = %{config | security_sasl_password:
redact(config.security_sasl_password)}
+
+ fields = sanitized |> Map.from_struct() |> Map.to_list()
+
+ container_doc(
+ "%Fluss.Config{",
+ fields,
+ "}",
+ opts,
+ fn {key, value}, opts ->
+ concat([Atom.to_string(key), ": ", to_doc(value, opts)])
+ end,
+ separator: ","
+ )
+ end
+
+ defp redact(nil), do: nil
+ defp redact(_), do: "[REDACTED]"
+end
diff --git a/bindings/elixir/native/fluss_nif/src/config.rs
b/bindings/elixir/native/fluss_nif/src/config.rs
index 79aabcd7..8c1bab51 100644
--- a/bindings/elixir/native/fluss_nif/src/config.rs
+++ b/bindings/elixir/native/fluss_nif/src/config.rs
@@ -31,6 +31,7 @@ pub enum NifNoKeyAssigner {
#[module = "Fluss.Config"]
pub struct NifConfig {
pub bootstrap_servers: String,
+ pub connect_timeout_ms: Option<u64>,
pub remote_file_download_thread_num: Option<u64>,
pub scanner_log_fetch_max_bytes: Option<i32>,
pub scanner_log_fetch_max_bytes_for_bucket: Option<i32>,
@@ -39,6 +40,10 @@ pub struct NifConfig {
pub scanner_log_max_poll_records: Option<u64>,
pub scanner_remote_log_prefetch_num: Option<u64>,
pub scanner_remote_log_read_concurrency: Option<u64>,
+ pub security_protocol: Option<String>,
+ pub security_sasl_mechanism: Option<String>,
+ pub security_sasl_password: Option<String>,
+ pub security_sasl_username: Option<String>,
pub writer_acks: Option<String>,
pub writer_batch_size: Option<i32>,
pub writer_batch_timeout_ms: Option<i64>,
@@ -59,6 +64,9 @@ impl NifConfig {
bootstrap_servers: self.bootstrap_servers,
..Config::default()
};
+ if let Some(timeout) = self.connect_timeout_ms {
+ config.connect_timeout_ms = timeout;
+ }
if let Some(n) = self.remote_file_download_thread_num {
config.remote_file_download_thread_num = n as usize;
}
@@ -83,6 +91,18 @@ impl NifConfig {
if let Some(n) = self.scanner_remote_log_read_concurrency {
config.scanner_remote_log_read_concurrency = n as usize;
}
+ if let Some(protocol) = self.security_protocol {
+ config.security_protocol = protocol;
+ }
+ if let Some(mechanism) = self.security_sasl_mechanism {
+ config.security_sasl_mechanism = mechanism;
+ }
+ if let Some(password) = self.security_sasl_password {
+ config.security_sasl_password = password;
+ }
+ if let Some(username) = self.security_sasl_username {
+ config.security_sasl_username = username;
+ }
if let Some(size) = self.writer_batch_size {
config.writer_batch_size = size;
}
diff --git a/bindings/elixir/test/config_test.exs
b/bindings/elixir/test/config_test.exs
index 2550a7e8..f4b8a11c 100644
--- a/bindings/elixir/test/config_test.exs
+++ b/bindings/elixir/test/config_test.exs
@@ -23,6 +23,14 @@ defmodule Fluss.ConfigTest do
assert config == %Fluss.Config{bootstrap_servers: "localhost:9123"}
end
+ test "set_connect_timeout_ms/2 sets the connect timeout" do
+ config =
+ Fluss.Config.new("localhost:9123")
+ |> Fluss.Config.set_connect_timeout_ms(30_000)
+
+ assert config.connect_timeout_ms == 30_000
+ end
+
test "set_remote_file_download_thread_num/2 sets the download thread num" do
config =
Fluss.Config.new("localhost:9123")
@@ -87,6 +95,54 @@ defmodule Fluss.ConfigTest do
assert config.scanner_remote_log_read_concurrency == 4
end
+ test "set_security_protocol/2 sets the security protocol" do
+ config =
+ Fluss.Config.new("localhost:9123")
+ |> Fluss.Config.set_security_protocol("sasl")
+
+ assert config.security_protocol == "sasl"
+ end
+
+ test "set_security_sasl_mechanism/2 sets the SASL mechanism" do
+ config =
+ Fluss.Config.new("localhost:9123")
+ |> Fluss.Config.set_security_sasl_mechanism("PLAIN")
+
+ assert config.security_sasl_mechanism == "PLAIN"
+ end
+
+ test "set_security_sasl_username/2 sets the SASL username" do
+ config =
+ Fluss.Config.new("localhost:9123")
+ |> Fluss.Config.set_security_sasl_username("admin")
+
+ assert config.security_sasl_username == "admin"
+ end
+
+ test "set_security_sasl_password/2 sets the SASL password" do
+ config =
+ Fluss.Config.new("localhost:9123")
+ |> Fluss.Config.set_security_sasl_password("secret")
+
+ assert config.security_sasl_password == "secret"
+ end
+
+ test "inspect/1 redacts security_sasl_password when set" do
+ config =
+ Fluss.Config.new("localhost:9123")
+ |> Fluss.Config.set_security_sasl_password("supersecret")
+
+ output = inspect(config)
+ refute output =~ "supersecret"
+ assert output =~ "[REDACTED]"
+ end
+
+ test "inspect/1 leaves nil security_sasl_password as nil" do
+ config = Fluss.Config.new("localhost:9123")
+ output = inspect(config)
+ assert output =~ "security_sasl_password: nil"
+ end
+
test "set_writer_acks/2 sets the acks value" do
config =
Fluss.Config.new("localhost:9123")
