This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new f7b878e137 [CELEBORN-2417] Pass CelebornConf properties through the
C++ FFI
f7b878e137 is described below
commit f7b878e137c582718094f44c7f8494cc8bd662f3
Author: Yu Gan <[email protected]>
AuthorDate: Tue Aug 11 20:11:09 2026 +0800
[CELEBORN-2417] Pass CelebornConf properties through the C++ FFI
### What changes were proposed in this pull request?
Replace the per-option parameters of `celeborn_ffi_create_client` with a
generic `(keys, values, num_props)` property array, and update the Rust wrapper
and examples accordingly.
C ABI:
```c
-celeborn_ffi_handle* celeborn_ffi_create_client(
- const char* app_id, size_t app_id_len,
- int32_t push_buffer_max_size,
- const char* codec, size_t codec_len,
- char** err_out);
+celeborn_ffi_handle* celeborn_ffi_create_client(
+ const char* app_id, size_t app_id_len,
+ const char* const* keys,
+ const char* const* values,
+ size_t num_props,
+ char** err_out);
```
Each pair is fed to `CelebornConf::registerProperty`, which already
validates key names and value formats. Null `keys`/`values` and null entries at
any index are rejected at the boundary with a descriptive error rather than
segfaulting.
On the Rust side, `Config::push_buffer_max_size` and
`Config::shuffle_compression_codec` are replaced by `properties: Vec<(String,
String)>` with a `set_property()` builder applied in insertion order:
```rust
let mut config = Config::new(app_id);
config.set_property("celeborn.client.shuffle.compression.codec", "ZSTD");
```
### Why are the changes needed?
Only `celeborn.client.push.buffer.max.size` and
`celeborn.client.shuffle.compression.codec` were reachable through the FFI.
Every further setting a binding wants to expose costs a C ABI break plus a
matching change in each downstream language wrapper, so in practice the rest of
`CelebornConf` is unavailable to non-Java clients. Passing properties through
verbatim means a binding picks up new configuration keys with no ABI change and
no code change at all.
It also removes a duplicated source of truth. The Rust wrapper carried its
own codec allow-list (`NONE` / `LZ4` / `ZSTD`) that the C++ side already owns,
so the two could drift; validation is now left to `registerProperty`. The
Rust-side checks that remain are the ones that are genuinely client-side: empty
`app_id`, non-positive port, empty property key, and interior NUL bytes in a
property string.
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [x] Yes
This is a deliberate break of the `celeborn_ffi_create_client` signature
and of the `Config` struct in the `celeborn-client` crate. Neither has a
released consumer outside this repository — the C ABI and the Rust crate are
both new and unpublished — and the two in-tree examples under `rust/examples/`
are updated in the same change. No Celeborn configuration key, Java API or wire
protocol is affected.
### How was this patch tested?
Unit tests in `rust/celeborn-client/src/lib.rs`.
`validate_rejects_unknown_codec` is dropped along with the allow-list it
covered; the following are added or reworked:
- `validate_accepts_valid_args`
- `validate_rejects_empty_property_key`
- `set_property_accumulates_native_keys` — properties are recorded in
insertion order
- `to_c_strings_rejects_interior_nul`
`cargo test -p celeborn-client` passes 6/6. `cargo clippy` reports one
`too_many_arguments` warning on `push_data`, which is pre-existing on `main`
and untouched here; `cargo fmt --check` reports two diffs in `shutdown`/`Drop`,
also pre-existing and untouched.
Note that no CI workflow currently builds or tests the `rust/` crates, so
these tests do not run in CI today — the `Celeborn Cpp Integration Test`
workflow compiles the C++ side including `CelebornFfi.cc`, but nothing
exercises the binding. If reviewers want Rust coverage wired into CI, I am
happy to do it in a follow-up.
Closes #3795 from yugan95/CELEBORN-2417.
Authored-by: Yu Gan <[email protected]>
Signed-off-by: Nicholas Jiang <[email protected]>
---
cpp/celeborn/ffi/CelebornFfi.cc | 27 +++++----
cpp/celeborn/ffi/CelebornFfi.h | 12 +++-
rust/celeborn-client-sys/src/lib.rs | 10 +++-
rust/celeborn-client/src/lib.rs | 106 +++++++++++++++++++++++++++---------
rust/examples/data_sum_reader.rs | 2 +-
rust/examples/data_sum_writer.rs | 2 +-
6 files changed, 110 insertions(+), 49 deletions(-)
diff --git a/cpp/celeborn/ffi/CelebornFfi.cc b/cpp/celeborn/ffi/CelebornFfi.cc
index 10a00081aa..331306a8cc 100644
--- a/cpp/celeborn/ffi/CelebornFfi.cc
+++ b/cpp/celeborn/ffi/CelebornFfi.cc
@@ -110,16 +110,16 @@ void celeborn_ffi_free_buffer(uint8_t* data) {
celeborn_ffi_handle* celeborn_ffi_create_client(
const char* app_id,
size_t app_id_len,
- int32_t push_buffer_max_size,
- const char* codec,
- size_t codec_len,
+ const char* const* keys,
+ const char* const* values,
+ size_t num_props,
char** err_out) {
if (app_id == nullptr && app_id_len > 0) {
set_error(err_out, "null pointer for argument 'app_id'");
return nullptr;
}
- if (codec == nullptr && codec_len > 0) {
- set_error(err_out, "null pointer for argument 'codec'");
+ if (num_props > 0 && (keys == nullptr || values == nullptr)) {
+ set_error(err_out, "null pointer for argument 'keys' or 'values'");
return nullptr;
}
try {
@@ -129,15 +129,14 @@ celeborn_ffi_handle* celeborn_ffi_create_client(
}
impl->conf = std::make_shared<celeborn::conf::CelebornConf>();
- if (push_buffer_max_size > 0) {
- impl->conf->registerProperty(
- celeborn::conf::CelebornConf::kClientPushBufferMaxSize,
- std::to_string(push_buffer_max_size) + "b");
- }
- if (codec_len > 0) {
- impl->conf->registerProperty(
- celeborn::conf::CelebornConf::kShuffleCompressionCodec,
- std::string(codec, codec_len));
+ for (size_t i = 0; i < num_props; i++) {
+ if (keys[i] == nullptr || values[i] == nullptr) {
+ set_error(
+ err_out,
+ "null key or value at property index " + std::to_string(i));
+ return nullptr;
+ }
+ impl->conf->registerProperty(keys[i], values[i]);
}
impl->endpoint =
diff --git a/cpp/celeborn/ffi/CelebornFfi.h b/cpp/celeborn/ffi/CelebornFfi.h
index e686113baf..962f75cc15 100644
--- a/cpp/celeborn/ffi/CelebornFfi.h
+++ b/cpp/celeborn/ffi/CelebornFfi.h
@@ -53,13 +53,19 @@ void celeborn_ffi_free_error(char* err);
// Releases a buffer returned via celeborn_ffi_read_partition_full.
void celeborn_ffi_free_buffer(uint8_t* data);
+// Creates a client configured from `num_props` key/value pairs, each a
+// NUL-terminated string naming a CelebornConf property (for example
+// "celeborn.client.push.buffer.max.size"). Passing the properties through
+// verbatim means bindings pick up new configuration keys without an ABI
+// change; unknown keys are rejected by CelebornConf::registerProperty.
+//
// Returns NULL on failure; in that case *err_out is set.
celeborn_ffi_handle* celeborn_ffi_create_client(
const char* app_id,
size_t app_id_len,
- int32_t push_buffer_max_size,
- const char* codec,
- size_t codec_len,
+ const char* const* keys,
+ const char* const* values,
+ size_t num_props,
char** err_out);
celeborn_ffi_status celeborn_ffi_setup_lifecycle_manager(
diff --git a/rust/celeborn-client-sys/src/lib.rs
b/rust/celeborn-client-sys/src/lib.rs
index 811e67fa7b..9cedd0f617 100644
--- a/rust/celeborn-client-sys/src/lib.rs
+++ b/rust/celeborn-client-sys/src/lib.rs
@@ -53,12 +53,16 @@ extern "C" {
pub fn celeborn_ffi_free_error(err: *mut c_char);
pub fn celeborn_ffi_free_buffer(data: *mut u8);
+ /// Creates a client configured from `num_props` NUL-terminated
+ /// key/value pairs, each naming a `CelebornConf` property. Passing the
+ /// properties through verbatim means new configuration keys need no
+ /// change to this signature.
pub fn celeborn_ffi_create_client(
app_id: *const c_char,
app_id_len: usize,
- push_buffer_max_size: i32,
- codec: *const c_char,
- codec_len: usize,
+ keys: *const *const c_char,
+ values: *const *const c_char,
+ num_props: usize,
err_out: *mut *mut c_char,
) -> *mut celeborn_ffi_handle;
diff --git a/rust/celeborn-client/src/lib.rs b/rust/celeborn-client/src/lib.rs
index 6b73cee79d..b9bb4f69d7 100644
--- a/rust/celeborn-client/src/lib.rs
+++ b/rust/celeborn-client/src/lib.rs
@@ -16,6 +16,7 @@
//! Rust-friendly wrapper around `celeborn-client-sys` (raw C ABI bindings
//! to `libceleborn_client.{so,dylib}`).
+use std::ffi::CString;
use std::marker::PhantomData;
use std::os::raw::c_char;
use std::ptr;
@@ -46,20 +47,27 @@ unsafe fn ffi_error(err: *mut c_char) -> Error {
/// Configuration for connecting to a Celeborn LifecycleManager.
pub struct Config {
pub app_id: String,
- /// Max push buffer size in bytes. 0 means use cpp default (64kB).
- pub push_buffer_max_size: i32,
- /// Compression codec: "NONE", "LZ4", or "ZSTD".
- pub shuffle_compression_codec: String,
+ /// Native `CelebornConf` properties, forwarded verbatim to the C++
+ /// client, e.g. `("celeborn.client.push.buffer.max.size", "64k")`.
+ /// Anything the C++ client understands can be set here without a change
+ /// to this crate; unknown keys are rejected at connect time.
+ pub properties: Vec<(String, String)>,
}
impl Config {
pub fn new(app_id: String) -> Self {
Self {
app_id,
- push_buffer_max_size: 0,
- shuffle_compression_codec: "NONE".to_string(),
+ properties: Vec::new(),
}
}
+
+ /// Sets a native `CelebornConf` property. Properties are applied in
+ /// insertion order, so a later entry for the same key wins.
+ pub fn set_property(&mut self, key: impl Into<String>, value: impl
Into<String>) -> &mut Self {
+ self.properties.push((key.into(), value.into()));
+ self
+ }
}
/// A Rust-friendly Celeborn shuffle client backed by the C++ implementation.
@@ -118,28 +126,44 @@ fn validate_connect_args(config: &Config, lm_port: i32)
-> Result<()> {
if lm_port <= 0 {
return Err(Error::InvalidArg("lm_port must be > 0"));
}
- let valid_codecs = ["NONE", "LZ4", "ZSTD"];
- if !valid_codecs.contains(&config.shuffle_compression_codec.as_str()) {
- return Err(Error::InvalidArg(
- "shuffle_compression_codec must be NONE, LZ4, or ZSTD",
- ));
+ if config.properties.iter().any(|(key, _)| key.is_empty()) {
+ return Err(Error::InvalidArg("property key is empty"));
}
Ok(())
}
+/// Copies borrowed strings into NUL-terminated `CString`s for the C ABI.
+fn to_c_strings<'a>(items: impl Iterator<Item = &'a str>) ->
Result<Vec<CString>> {
+ items
+ .map(|item| {
+ CString::new(item)
+ .map_err(|_| Error::InvalidArg("property contains an interior
NUL byte"))
+ })
+ .collect()
+}
+
impl ShuffleClient {
/// Connect to a running LifecycleManager at `lm_host:lm_port`.
pub fn connect(config: Config, lm_host: &str, lm_port: i32) ->
Result<Self> {
validate_connect_args(&config, lm_port)?;
+ // The C++ side expects NUL-terminated strings, so the properties are
+ // copied into CStrings. `keys`/`values` own those buffers and must
+ // stay alive for the duration of the call below; `key_ptrs`/
+ // `value_ptrs` only borrow from them.
+ let keys = to_c_strings(config.properties.iter().map(|(key, _)|
key.as_str()))?;
+ let values = to_c_strings(config.properties.iter().map(|(_, value)|
value.as_str()))?;
+ let key_ptrs: Vec<*const c_char> = keys.iter().map(|key|
key.as_ptr()).collect();
+ let value_ptrs: Vec<*const c_char> = values.iter().map(|value|
value.as_ptr()).collect();
+
let mut err: *mut c_char = ptr::null_mut();
let handle = unsafe {
sys::celeborn_ffi_create_client(
config.app_id.as_ptr() as *const c_char,
config.app_id.len(),
- config.push_buffer_max_size,
- config.shuffle_compression_codec.as_ptr() as *const c_char,
- config.shuffle_compression_codec.len(),
+ key_ptrs.as_ptr(),
+ value_ptrs.as_ptr(),
+ key_ptrs.len(),
&mut err,
)
};
@@ -433,30 +457,27 @@ impl<'client> Drop for PartitionReader<'client> {
mod tests {
use super::*;
- fn config_with(app_id: &str, codec: &str) -> Config {
- let mut config = Config::new(app_id.to_string());
- config.shuffle_compression_codec = codec.to_string();
- config
+ fn config_with(app_id: &str) -> Config {
+ Config::new(app_id.to_string())
}
#[test]
- fn validate_accepts_supported_codecs() {
- for codec in ["NONE", "LZ4", "ZSTD"] {
- let config = config_with("app", codec);
- assert!(validate_connect_args(&config, 39099).is_ok());
- }
+ fn validate_accepts_valid_args() {
+ let mut config = config_with("app");
+ config.set_property("celeborn.client.shuffle.compression.codec",
"LZ4");
+ assert!(validate_connect_args(&config, 39099).is_ok());
}
#[test]
fn validate_rejects_empty_app_id() {
- let config = config_with("", "NONE");
+ let config = config_with("");
let err = validate_connect_args(&config, 39099).unwrap_err();
assert!(matches!(err, Error::InvalidArg("app_id is empty")));
}
#[test]
fn validate_rejects_non_positive_port() {
- let config = config_with("app", "NONE");
+ let config = config_with("app");
assert!(matches!(
validate_connect_args(&config, 0).unwrap_err(),
Error::InvalidArg("lm_port must be > 0")
@@ -468,10 +489,41 @@ mod tests {
}
#[test]
- fn validate_rejects_unknown_codec() {
- let config = config_with("app", "GZIP");
+ fn validate_rejects_empty_property_key() {
+ let mut config = config_with("app");
+ config.set_property("", "value");
assert!(matches!(
validate_connect_args(&config, 39099).unwrap_err(),
+ Error::InvalidArg("property key is empty")
+ ));
+ }
+
+ #[test]
+ fn set_property_accumulates_native_keys() {
+ let mut config = config_with("app");
+ config
+ .set_property("celeborn.client.push.buffer.max.size", "64k")
+ .set_property("celeborn.client.shuffle.compression.codec", "ZSTD");
+ assert_eq!(
+ config.properties,
+ vec![
+ (
+ "celeborn.client.push.buffer.max.size".to_string(),
+ "64k".to_string()
+ ),
+ (
+ "celeborn.client.shuffle.compression.codec".to_string(),
+ "ZSTD".to_string()
+ ),
+ ]
+ );
+ }
+
+ #[test]
+ fn to_c_strings_rejects_interior_nul() {
+ assert!(to_c_strings(["ok"].into_iter()).is_ok());
+ assert!(matches!(
+ to_c_strings(["ba\0d"].into_iter()).unwrap_err(),
Error::InvalidArg(_)
));
}
diff --git a/rust/examples/data_sum_reader.rs b/rust/examples/data_sum_reader.rs
index 18c34a26fd..d2a81650ca 100644
--- a/rust/examples/data_sum_reader.rs
+++ b/rust/examples/data_sum_reader.rs
@@ -62,7 +62,7 @@ fn main() {
);
let mut config = Config::new(app_id);
- config.shuffle_compression_codec = compress_codec;
+ config.set_property("celeborn.client.shuffle.compression.codec",
compress_codec);
let client =
Arc::new(ShuffleClient::connect(config, lm_host,
lm_port).expect("Failed to connect to LM"));
diff --git a/rust/examples/data_sum_writer.rs b/rust/examples/data_sum_writer.rs
index 5d7780bd50..a3213d642b 100644
--- a/rust/examples/data_sum_writer.rs
+++ b/rust/examples/data_sum_writer.rs
@@ -64,7 +64,7 @@ fn main() {
);
let mut config = Config::new(app_id);
- config.shuffle_compression_codec = compress_codec;
+ config.set_property("celeborn.client.shuffle.compression.codec",
compress_codec);
let client =
Arc::new(ShuffleClient::connect(config, lm_host,
lm_port).expect("Failed to connect to LM"));
