[rust] Check in generated protobuf code and drop the protoc build requirement

.github/workflows/build_and_test_rust.yml

@@ -51,11 +51,6 @@ jobs:
     steps:
       - uses: actions/checkout@v6
 
-      - name: Install protoc
-        uses: arduino/setup-protoc@c65c819552d16ad3c9b72d9dfd5ba5237b9c906b # v3.0.0
-        with:
-          repo-token: ${{ secrets.GITHUB_TOKEN }}
-
       - name: Rust Cache
         uses: Swatinem/rust-cache@c19371144df3bb44fab255c43d04cbc2ab54d1c4 # v2.9.1
 
@@ -75,11 +70,6 @@ jobs:
     steps:
       - uses: actions/checkout@v6
 
-      - name: Install protoc
-        uses: arduino/setup-protoc@c65c819552d16ad3c9b72d9dfd5ba5237b9c906b # v3.0.0
-        with:
-          repo-token: ${{ secrets.GITHUB_TOKEN }}
-
       - name: Rust Cache
         uses: Swatinem/rust-cache@c19371144df3bb44fab255c43d04cbc2ab54d1c4 # v2.9.1
 
@@ -88,3 +78,23 @@ jobs:
         env:
           RUST_LOG: DEBUG
           RUST_BACKTRACE: full
+
+  proto-up-to-date:
+    timeout-minutes: 15
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Install protoc
+        uses: arduino/setup-protoc@c65c819552d16ad3c9b72d9dfd5ba5237b9c906b # v3.0.0
+        with:
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Rust Cache
+        uses: Swatinem/rust-cache@c19371144df3bb44fab255c43d04cbc2ab54d1c4 # v2.9.1
+
+      - name: Regenerate protobuf code
+        run: ./crates/fluss/regen.sh
+
+      - name: Verify the checked-in code is up to date
+        run: git diff --exit-code

.github/workflows/check_license_and_formatting.yml

@@ -53,11 +53,6 @@ jobs:
       - name: Check dependency licenses (Apache-compatible)
         run: cargo deny check licenses
 
-      - name: Install protoc
-        uses: arduino/setup-protoc@c65c819552d16ad3c9b72d9dfd5ba5237b9c906b # v3.0.0
-        with:
-          repo-token: ${{ secrets.GITHUB_TOKEN }}
-
       - name: Rust Cache
         uses: Swatinem/rust-cache@c19371144df3bb44fab255c43d04cbc2ab54d1c4 # v2.9.1
 

Cargo.toml

@@ -28,7 +28,7 @@ keywords = ["fluss", "streaming-storage", "datalake"]
 
 [workspace]
 resolver = "2"
-members = ["crates/fluss", "crates/fluss-test-cluster", "crates/examples", "bindings/python", "bindings/cpp", "bindings/elixir/native/fluss_nif"]
+members = ["crates/fluss", "crates/fluss/gen", "crates/fluss-test-cluster", "crates/examples", "bindings/python", "bindings/cpp", "bindings/elixir/native/fluss_nif"]
 
 [workspace.dependencies]
 fluss = { package = "fluss-rs", version = "0.2.0", path = "crates/fluss", features = ["storage-all"] }

DEVELOPMENT.md

@@ -4,16 +4,16 @@ Welcome to the development guide of `fluss-rust`! This project builds `fluss-rus
 
 ## Pre-requisites
 
-- protobuf
 - rust
 
 You can install these using your favourite package / version manager. Example installation using mise:
 
 ```bash
-mise install protobuf
 mise install rust
 ```
 
+`protobuf` (the `protoc` compiler) is only needed when regenerating the protobuf code after changing `crates/fluss/src/proto/fluss_api.proto` (see `crates/fluss/regen.sh`). Regular builds use the checked-in `crates/fluss/src/proto/proto.rs`.
+
 ## IDE Setup
 
 We recommend [RustRover](https://www.jetbrains.com/rust/) IDE to work with fluss-rust code base.

crates/fluss/Cargo.toml

@@ -86,6 +86,3 @@ fluss-test-cluster = { path = "../fluss-test-cluster" }
 # workspace `full` feature set; enable it for tests so the
 # `last_poll_seconds_ago` ticker loop can be driven deterministically.
 tokio = { workspace = true, features = ["test-util"] }
-
-[build-dependencies]
-prost-build = "0.14"

crates/fluss/gen/Cargo.toml

@@ -0,0 +1,33 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "gen"
+description = "Code generation for fluss-rs"
+version = "0.1.0"
+edition = { workspace = true }
+rust-version = { workspace = true }
+authors = { workspace = true }
+homepage = { workspace = true }
+repository = { workspace = true }
+license = { workspace = true }
+publish = false
+
+[dependencies]
+# Pin the prost-build version so regenerating does not churn the checked-in
+# src/proto/proto.rs. Bump deliberately, then rerun ../regen.sh.
+prost-build = "0.14.3"

crates/fluss/gen/src/main.rs

@@ -0,0 +1,66 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::error::Error;
+use std::fs;
+use std::path::Path;
+
+const HEADER: &str = "\
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// \"License\"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// This file was generated from src/proto/fluss_api.proto by crates/fluss/regen.sh,
+// and should not be edited by hand.
+
+";
+
+fn main() -> Result<(), Box<dyn Error>> {
+    let fluss_dir = Path::new(env!("CARGO_MANIFEST_DIR"))
+        .parent()
+        .ok_or("the gen crate must live inside crates/fluss")?;
+    let proto_dir = fluss_dir.join("src/proto");
+
+    let mut config = prost_build::Config::new();
+    config.bytes([
+        ".proto.PbProduceLogReqForBucket.records",
+        ".proto.PbPutKvReqForBucket.records",
+        ".proto.PbLookupReqForBucket.keys",
+        ".proto.PbPrefixLookupReqForBucket.keys",
+        ".proto.ScanKvResponse.records",
+    ]);
+    config.out_dir(&proto_dir);
+    config.compile_protos(&[proto_dir.join("fluss_api.proto")], &[&proto_dir])?;
+
+    let generated_path = proto_dir.join("proto.rs");
+    let generated = fs::read_to_string(&generated_path)?;
+    fs::write(&generated_path, format!("{HEADER}{generated}"))?;
+    Ok(())
+}

crates/fluss/regen.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Regenerates src/proto/proto.rs from src/proto/fluss_api.proto.
+# Requires protoc on PATH (or the PROTOC env var pointing at a protoc binary).
+
+set -euo pipefail
+
+SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
+cd "$SCRIPT_DIR" && cargo run --manifest-path gen/Cargo.toml

crates/fluss/src/client/write/accumulator.rs

@@ -877,7 +877,7 @@ impl RecordAccumulator {
         self.memory_limiter.close();
         // Complete batches still in deques (not yet drained).
         for mut entry in self.write_batches.iter_mut() {
-            for (_bucket_id, deque) in entry.value_mut().batches.iter_mut() {
+            for deque in entry.value_mut().batches.values_mut() {
                 let mut dq = deque.lock();
                 while let Some(batch) = dq.pop_front() {
                     batch.complete(Err(error.clone()));
@@ -928,7 +928,7 @@ impl RecordAccumulator {
 
     pub fn has_undrained(&self) -> bool {
         for entry in self.write_batches.iter() {
-            for (_, batch_deque) in entry.value().batches.iter() {
+            for batch_deque in entry.value().batches.values() {
                 if !batch_deque.lock().is_empty() {
                     return true;
                 }

crates/fluss/src/lib.rs

@@ -149,5 +149,5 @@ pub type BucketId = i32;
 pub(crate) mod proto {
     // Generated; not every 1.x message is wired up to a caller yet.
     #![allow(dead_code)]
-    include!(concat!(env!("OUT_DIR"), "/proto.rs"));
+    include!("proto/proto.rs");
 }