Skip to content

Commit 669cec7

Browse files
gnuhpcwarmbuptclaude
authored
[rust] Add RPC message wrappers for extended operations (#630)
* [rust] Add RPC message wrappers for extended operations Add message wrappers for the remaining 1.x RPC APIs: - KV snapshot lifecycle: acquire/release/drop lease, list, metadata, latest snapshots, lake snapshot - Server management: add/remove server tag, rebalance + progress + cancel, get cluster health, list remote log manifests - Producer offsets: register/get/delete - ScanKv (API 1061): full KV-table bucket scan request/response * Add GoalType and ServerTag domain enums for RPC wrappers Replace raw i32 with proper Rust enums matching the Java Fluss definitions: GoalType (ReplicaDistribution/LeaderDistribution/RackAware) and ServerTag (PermanentOffline/TemporaryOffline). Addresses reviewer feedback on PR #630. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * Apply TableId/PartitionId/BucketId type aliases to RPC messages Use the i64/i32 type aliases from lib.rs instead of raw primitives in the new RPC message wrappers introduced by this PR. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> --------- Co-authored-by: warmbupt <warmbupt@gmail.com> Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
1 parent ee6a5ac commit 669cec7

27 files changed

Lines changed: 1315 additions & 4 deletions

‎crates/fluss/src/metadata/goal_type.rs‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
use crate::error::{Error, Result};
19+
20+
/// Mirrors Java `org.apache.fluss.cluster.rebalance.GoalType`.
21+
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22+
pub enum GoalType {
23+
ReplicaDistribution,
24+
LeaderDistribution,
25+
RackAware,
26+
}
27+
28+
impl GoalType {
29+
pub fn to_i32(self) -> i32 {
30+
match self {
31+
Self::ReplicaDistribution => 0,
32+
Self::LeaderDistribution => 1,
33+
Self::RackAware => 2,
34+
}
35+
}
36+
37+
pub fn try_from_i32(value: i32) -> Result<Self> {
38+
match value {
39+
0 => Ok(Self::ReplicaDistribution),
40+
1 => Ok(Self::LeaderDistribution),
41+
2 => Ok(Self::RackAware),
42+
_ => Err(Error::IllegalArgument {
43+
message: format!("Unsupported GoalType: {value}"),
44+
}),
45+
}
46+
}
47+
}
48+
49+
#[cfg(test)]
50+
mod tests {
51+
use super::*;
52+
53+
#[test]
54+
fn test_goal_type_roundtrip() {
55+
for goal in [
56+
GoalType::ReplicaDistribution,
57+
GoalType::LeaderDistribution,
58+
GoalType::RackAware,
59+
] {
60+
assert_eq!(GoalType::try_from_i32(goal.to_i32()).unwrap(), goal);
61+
}
62+
}
63+
64+
#[test]
65+
fn test_goal_type_unknown() {
66+
assert!(GoalType::try_from_i32(99).is_err());
67+
}
68+
}

‎crates/fluss/src/metadata/kv_snapshot_lease.rs‎

Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,118 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
use crate::proto::{PbKvSnapshotLeaseForBucket, PbKvSnapshotLeaseForTable};
19+
20+
/// One bucket's slot in a KV-snapshot lease request.
21+
#[derive(Debug, Clone, PartialEq, Eq)]
22+
pub struct KvSnapshotLeaseForBucket {
23+
pub partition_id: Option<i64>,
24+
pub bucket_id: i32,
25+
pub snapshot_id: i64,
26+
}
27+
28+
impl KvSnapshotLeaseForBucket {
29+
pub fn to_pb(&self) -> PbKvSnapshotLeaseForBucket {
30+
PbKvSnapshotLeaseForBucket {
31+
partition_id: self.partition_id,
32+
bucket_id: self.bucket_id,
33+
snapshot_id: self.snapshot_id,
34+
}
35+
}
36+
37+
pub fn from_pb(pb: &PbKvSnapshotLeaseForBucket) -> Self {
38+
Self {
39+
partition_id: pb.partition_id,
40+
bucket_id: pb.bucket_id,
41+
snapshot_id: pb.snapshot_id,
42+
}
43+
}
44+
}
45+
46+
/// All the buckets of a single table that should be leased together.
47+
#[derive(Debug, Clone, PartialEq, Eq)]
48+
pub struct KvSnapshotLeaseForTable {
49+
pub table_id: i64,
50+
pub bucket_snapshots: Vec<KvSnapshotLeaseForBucket>,
51+
}
52+
53+
impl KvSnapshotLeaseForTable {
54+
pub fn to_pb(&self) -> PbKvSnapshotLeaseForTable {
55+
PbKvSnapshotLeaseForTable {
56+
table_id: self.table_id,
57+
bucket_snapshots: self
58+
.bucket_snapshots
59+
.iter()
60+
.map(KvSnapshotLeaseForBucket::to_pb)
61+
.collect(),
62+
}
63+
}
64+
65+
pub fn from_pb(pb: &PbKvSnapshotLeaseForTable) -> Self {
66+
Self {
67+
table_id: pb.table_id,
68+
bucket_snapshots: pb
69+
.bucket_snapshots
70+
.iter()
71+
.map(KvSnapshotLeaseForBucket::from_pb)
72+
.collect(),
73+
}
74+
}
75+
}
76+
77+
#[cfg(test)]
78+
mod tests {
79+
use super::*;
80+
81+
#[test]
82+
fn test_kv_snapshot_lease_for_bucket_roundtrip() {
83+
for b in [
84+
KvSnapshotLeaseForBucket {
85+
partition_id: None,
86+
bucket_id: 0,
87+
snapshot_id: 10,
88+
},
89+
KvSnapshotLeaseForBucket {
90+
partition_id: Some(42),
91+
bucket_id: 3,
92+
snapshot_id: 99,
93+
},
94+
] {
95+
assert_eq!(KvSnapshotLeaseForBucket::from_pb(&b.to_pb()), b);
96+
}
97+
}
98+
99+
#[test]
100+
fn test_kv_snapshot_lease_for_table_roundtrip() {
101+
let t = KvSnapshotLeaseForTable {
102+
table_id: 7,
103+
bucket_snapshots: vec![
104+
KvSnapshotLeaseForBucket {
105+
partition_id: None,
106+
bucket_id: 0,
107+
snapshot_id: 10,
108+
},
109+
KvSnapshotLeaseForBucket {
110+
partition_id: Some(42),
111+
bucket_id: 1,
112+
snapshot_id: 11,
113+
},
114+
],
115+
};
116+
assert_eq!(KvSnapshotLeaseForTable::from_pb(&t.to_pb()), t);
117+
}
118+
}

‎crates/fluss/src/metadata/mod.rs‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,13 @@ mod config;
2020
mod data_lake_format;
2121
mod database;
2222
mod datatype;
23+
mod goal_type;
2324
mod json_serde;
25+
mod kv_snapshot_lease;
2426
mod partition;
27+
mod producer_offsets;
2528
mod schema_util;
29+
mod server_tag;
2630
mod table;
2731
mod table_change;
2832
mod table_stats;
@@ -32,9 +36,13 @@ pub use config::*;
3236
pub use data_lake_format::*;
3337
pub use database::*;
3438
pub use datatype::*;
39+
pub use goal_type::*;
3540
pub use json_serde::*;
41+
pub use kv_snapshot_lease::*;
3642
pub use partition::*;
43+
pub use producer_offsets::*;
3744
pub(crate) use schema_util::{UNEXIST_MAPPING, index_mapping};
45+
pub use server_tag::*;
3846
pub use table::*;
3947
pub use table_change::*;
4048
pub use table_stats::*;

‎crates/fluss/src/metadata/producer_offsets.rs‎

Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,118 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
use crate::proto::{PbBucketOffset, PbProducerTableOffsets};
19+
20+
/// Per-bucket producer log-end offset.
21+
#[derive(Debug, Clone, PartialEq, Eq)]
22+
pub struct BucketOffset {
23+
pub partition_id: Option<i64>,
24+
pub bucket_id: i32,
25+
pub log_end_offset: Option<i64>,
26+
}
27+
28+
impl BucketOffset {
29+
pub fn to_pb(&self) -> PbBucketOffset {
30+
PbBucketOffset {
31+
partition_id: self.partition_id,
32+
bucket_id: self.bucket_id,
33+
log_end_offset: self.log_end_offset,
34+
}
35+
}
36+
37+
pub fn from_pb(pb: &PbBucketOffset) -> Self {
38+
Self {
39+
partition_id: pb.partition_id,
40+
bucket_id: pb.bucket_id,
41+
log_end_offset: pb.log_end_offset,
42+
}
43+
}
44+
}
45+
46+
/// All bucket offsets of a single table belonging to one producer.
47+
#[derive(Debug, Clone, PartialEq, Eq)]
48+
pub struct ProducerTableOffsets {
49+
pub table_id: i64,
50+
pub bucket_offsets: Vec<BucketOffset>,
51+
}
52+
53+
impl ProducerTableOffsets {
54+
pub fn to_pb(&self) -> PbProducerTableOffsets {
55+
PbProducerTableOffsets {
56+
table_id: self.table_id,
57+
bucket_offsets: self
58+
.bucket_offsets
59+
.iter()
60+
.map(BucketOffset::to_pb)
61+
.collect(),
62+
}
63+
}
64+
65+
pub fn from_pb(pb: &PbProducerTableOffsets) -> Self {
66+
Self {
67+
table_id: pb.table_id,
68+
bucket_offsets: pb
69+
.bucket_offsets
70+
.iter()
71+
.map(BucketOffset::from_pb)
72+
.collect(),
73+
}
74+
}
75+
}
76+
77+
#[cfg(test)]
78+
mod tests {
79+
use super::*;
80+
81+
#[test]
82+
fn test_bucket_offset_roundtrip() {
83+
for b in [
84+
BucketOffset {
85+
partition_id: None,
86+
bucket_id: 0,
87+
log_end_offset: None,
88+
},
89+
BucketOffset {
90+
partition_id: Some(42),
91+
bucket_id: 3,
92+
log_end_offset: Some(1234),
93+
},
94+
] {
95+
assert_eq!(BucketOffset::from_pb(&b.to_pb()), b);
96+
}
97+
}
98+
99+
#[test]
100+
fn test_producer_table_offsets_roundtrip() {
101+
let t = ProducerTableOffsets {
102+
table_id: 5,
103+
bucket_offsets: vec![
104+
BucketOffset {
105+
partition_id: None,
106+
bucket_id: 0,
107+
log_end_offset: Some(100),
108+
},
109+
BucketOffset {
110+
partition_id: Some(7),
111+
bucket_id: 1,
112+
log_end_offset: None,
113+
},
114+
],
115+
};
116+
assert_eq!(ProducerTableOffsets::from_pb(&t.to_pb()), t);
117+
}
118+
}

‎crates/fluss/src/metadata/server_tag.rs‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
use crate::error::{Error, Result};
19+
20+
/// Mirrors Java `org.apache.fluss.cluster.rebalance.ServerTag`.
21+
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22+
pub enum ServerTag {
23+
PermanentOffline,
24+
TemporaryOffline,
25+
}
26+
27+
impl ServerTag {
28+
pub fn to_i32(self) -> i32 {
29+
match self {
30+
Self::PermanentOffline => 0,
31+
Self::TemporaryOffline => 1,
32+
}
33+
}
34+
35+
pub fn try_from_i32(value: i32) -> Result<Self> {
36+
match value {
37+
0 => Ok(Self::PermanentOffline),
38+
1 => Ok(Self::TemporaryOffline),
39+
_ => Err(Error::IllegalArgument {
40+
message: format!("Unsupported ServerTag: {value}"),
41+
}),
42+
}
43+
}
44+
}
45+
46+
#[cfg(test)]
47+
mod tests {
48+
use super::*;
49+
50+
#[test]
51+
fn test_server_tag_roundtrip() {
52+
for tag in [ServerTag::PermanentOffline, ServerTag::TemporaryOffline] {
53+
assert_eq!(ServerTag::try_from_i32(tag.to_i32()).unwrap(), tag);
54+
}
55+
}
56+
57+
#[test]
58+
fn test_server_tag_unknown() {
59+
assert!(ServerTag::try_from_i32(99).is_err());
60+
}
61+
}

0 commit comments

Comments
 (0)