Update the fork

.github/CODEOWNERS

@@ -0,0 +1,3 @@
+# These owners will be the default owners for everything in
+# the repo. Unless a later match takes precedence
+*       @yhl25 @vigith

.github/workflows/build-push.yml

@@ -19,7 +19,8 @@ jobs:
           "examples/map-cat", "examples/map-tickgen-serde", "examples/mapt-event-time-filter",
           "examples/reduce-counter", "examples/sideinput", "examples/sideinput/udf",
           "examples/simple-source", "examples/sink-log", "examples/source-transformer-now",
-          "examples/batchmap-flatmap", "examples/batchmap-cat",
+          "examples/batchmap-flatmap", "examples/batchmap-cat", "examples/map-even-odd",
+          "examples/session-counter", "examples/stream-sorter","examples/flatmap-stream",
         ]
 
     steps:

.github/workflows/ci.yaml

@@ -25,6 +25,7 @@ jobs:
       - name: Install Rust
         uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
+          components: rustfmt, clippy
           cache-workspaces: |
             . -> target
 
@@ -34,7 +35,7 @@ jobs:
           echo "SCCACHE_GHA_ENABLED=true" >> $GITHUB_ENV
 
       - name: Run sccache-cache
-        uses: mozilla-actions/sccache-action@v0.0.5
+        uses: mozilla-actions/sccache-action@v0.0.9
 
       - name: Install dependencies
         run: sudo apt-get install -y protobuf-compiler

.rustfmt.toml

@@ -1 +1 @@
-edition = "2021"
+edition = "2024"

Cargo.toml

@@ -5,4 +5,27 @@ default-members = ["numaflow"]
 resolver = "2"
 
 [workspace.package]
-rust-version = "1.84"
+rust-version = "1.85.0"
+edition = "2024"
+
+[workspace.dependencies]
+tonic = "0.14.2"
+tonic-prost = "0.14.2"
+tonic-prost-build = "0.14.2"
+tonic-types = "0.14.2"
+prost = "0.14.1"
+prost-types = "0.14.1"
+tokio = { version = "1.0", features = ["macros", "rt-multi-thread", "signal"] }
+tokio-util = "0.7.16"
+tokio-stream = { version = "0.1.17", features = ["net"] }
+serde = { version = "1.0.228", features = ["derive"] }
+serde_json = "1.0.145"
+chrono = "0.4.42"
+tracing = "0.1.41"
+tracing-subscriber = "0.3"
+uuid = { version = "1.18.1", features = ["v4"] }
+thiserror = "2.0.17"
+hyper-util = "0.1.17"
+notify = "8.2.0"
+tempfile = "3.23.0"
+tower = { version = "0.5.2", features = ["util"] }

Makefile

@@ -1,4 +1,8 @@
-.PHONY: lint fmt test-fmt clippy test codegen
+.PHONY: lint fmt test-fmt clippy test codegen build clean-proto codegen-clean
+
+.PHONY: build
+build:
+	cargo build --workspace
 
 fmt:
 	cargo fmt --all
@@ -12,14 +16,34 @@ test-fmt:
 
 .PHONY: clippy
 clippy:
-	cargo clippy --workspace -- -D warnings
+	cargo clippy --workspace -- -D warnings -A clippy::module_inception
 
 # run cargo test on the repository root
 .PHONY: test
 test:
+	@echo "Running tests"
 	cargo test --workspace
+	@echo "Running panic tests sequentially..."
+	cargo test --workspace --features test-panic sink::tests::sink_panic
+	cargo test --workspace --features test-panic sourcetransform::tests::source_transformer_panic
+	cargo test --workspace --features test-panic map::tests::map_server_panic
+	cargo test --workspace --features test-panic mapstream::tests::map_stream_server_panic
+	cargo test --workspace --features test-panic batchmap::tests::batchmap_panic
+	cargo test --workspace --features test-panic reduce::tests::panic_tests::panic_in_reduce
+	cargo test --workspace --features test-panic reduce::tests::panic_tests::panic_with_multiple_keys
+	cargo test --workspace --features test-panic session_reduce::tests::panic_tests::test_panic_in_session_reduce
+	cargo test --workspace --features test-panic accumulator::tests::panic_tests::test_panic_in_accumulate
+	cargo test --workspace --features test-panic shared::panic_tests::test_panic_hook_functionality
 
 .PHONY: codegen
 codegen:
 	# Change timestamps so that tonic_build code generation will always be triggered.
-	cd numaflow && mkdir -p src/servers && touch proto/* && PROTO_CODE_GEN=1 cargo build
+	cd numaflow && touch proto/* && PROTO_CODE_GEN=1 cargo build
+
+.PHONY: clean-proto
+clean-proto:
+	# Clean generated proto files
+	cd numaflow && rm -rf src/generated
+
+.PHONY: codegen-clean
+codegen-clean: clean-proto codegen

examples/all-sinks/.dockerignore

@@ -0,0 +1 @@
+target/

examples/all-sinks/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "all-sinks"
+version = "0.1.0"
+edition.workspace = true
+rust-version.workspace = true
+
+[dependencies]
+rand = "0.9.2"
+tonic.workspace = true
+tokio.workspace = true
+tracing-subscriber.workspace = true
+numaflow = { path = "../../numaflow" }

examples/all-sinks/Dockerfile

@@ -0,0 +1,20 @@
+FROM rust:1.85-bullseye AS build
+
+RUN apt-get update
+RUN apt-get install protobuf-compiler -y
+
+WORKDIR /numaflow-rs
+COPY ./ ./
+WORKDIR /numaflow-rs/examples/all-sinks
+
+# build for release
+RUN cargo build --release
+
+# our final base
+FROM debian:bullseye AS all-sinks
+
+# copy the build artifact from the build stage
+COPY --from=build /numaflow-rs/target/release/all-sinks .
+
+# set the startup command to run your binary
+CMD ["./all-sinks"]

examples/all-sinks/Makefile

@@ -0,0 +1,16 @@
+TAG ?= stable
+PUSH ?= false
+IMAGE_REGISTRY = quay.io/numaio/numaflow-rs/all-sinks:${TAG}
+DOCKER_FILE_PATH = examples/all-sinks/Dockerfile
+
+.PHONY: update
+update:
+	cargo check
+	cargo update
+
+.PHONY: image
+image: update
+	cd ../../ && docker build \
+	-f ${DOCKER_FILE_PATH}  \
+	-t ${IMAGE_REGISTRY} .
+	@if [ "$(PUSH)" = "true" ]; then docker push ${IMAGE_REGISTRY}; fi

examples/all-sinks/manifests/all-sinks.yaml

@@ -0,0 +1,33 @@
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: all-sinks
+spec:
+  vertices:
+    - name: in
+      source:
+        # A self data generating source
+        generator:
+          rpu: 300
+          duration: 1s
+          keyCount: 5
+          value: 5
+    - name: out
+      sink:
+        udsink:
+          container:
+            image: quay.io/numaio/numaflow-rs/all-sinks:stable
+            imagePullPolicy: IfNotPresent
+        fallback:
+          udsink:
+            container:
+              image: quay.io/numaio/numaflow-rs/sink-log:stable
+              imagePullPolicy: IfNotPresent
+        onSuccess:
+          udsink:
+            container:
+              image: quay.io/numaio/numaflow-rs/sink-log:stable
+              imagePullPolicy: IfNotPresent
+  edges:
+    - from: in
+      to: out

examples/all-sinks/src/main.rs

@@ -0,0 +1,60 @@
+use numaflow::sink::KeyValueGroup;
+use numaflow::sink::{self, Message, Response, SinkRequest};
+use rand::Rng;
+use std::collections::HashMap;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    tracing_subscriber::fmt::init();
+    sink::Server::new(SinkHandler).start().await
+}
+
+struct SinkHandler;
+
+#[tonic::async_trait]
+impl sink::Sinker for SinkHandler {
+    async fn sink(&self, mut input: tokio::sync::mpsc::Receiver<SinkRequest>) -> Vec<Response> {
+        let mut responses: Vec<Response> = Vec::new();
+
+        while let Some(datum) = input.recv().await {
+            let response = match primary_sink_write_status() {
+                // Writing to primary sink was successful, we want to send a message to on_success sink
+                // The message sent to on_success sink can be different from the original message
+                true => {
+                    // build user metadata for on_success sink message payload
+                    let user_metadata = HashMap::from([(
+                        String::from("key1"),
+                        KeyValueGroup::from(HashMap::from([(
+                            datum.id.clone(),
+                            datum.value.clone(),
+                        )])),
+                    )]);
+                    Response::on_success(
+                        datum.id,
+                        // optional message, send original message to on_success sink in case `None` is provided
+                        Message::new(datum.value) // required value
+                            .with_keys(vec!["key1".to_string()]) // optional keys
+                            .with_user_metadata(user_metadata) // optional metadata
+                            .build(),
+                    )
+                }
+                // Writing to primary sink was not successful, we want to send a message to fallback sink
+                // The message sent to the fallback sink will be the original message
+                false => Response::fallback(datum.id),
+            };
+
+            // return the responses
+            responses.push(response);
+        }
+
+        responses
+    }
+}
+
+/// Get status of writing to primary sink
+/// true: write successful
+/// false: write failed
+fn primary_sink_write_status() -> bool {
+    // dummy logic to simulate write status
+    rand::rng().random_range(0..=1) == 1
+}

examples/batchmap-cat/Cargo.toml

@@ -1,9 +1,10 @@
 [package]
 name = "batchmap-cat"
 version = "0.1.0"
-edition = "2021"
+edition.workspace = true
+rust-version.workspace = true
 
 [dependencies]
-tonic = "0.13.0"
-tokio = { version = "1.0", features = ["macros", "rt-multi-thread"] }
+tonic.workspace = true
+tokio.workspace = true
 numaflow = { path = "../../numaflow" }

examples/batchmap-cat/Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.84-bullseye AS build
+FROM rust:1.85-bullseye AS build
 
 RUN apt-get update
 RUN apt-get install protobuf-compiler -y

examples/batchmap-flatmap/Cargo.toml

@@ -1,9 +1,10 @@
 [package]
 name = "batchmap-flatmap"
 version = "0.1.0"
-edition = "2021"
+edition.workspace = true
+rust-version.workspace = true
 
 [dependencies]
-tonic = "0.13.0"
-tokio = { version = "1.0", features = ["macros", "rt-multi-thread"] }
+tonic.workspace = true
+tokio.workspace = true
 numaflow = { path = "../../numaflow" }

examples/batchmap-flatmap/Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.84-bullseye AS build
+FROM rust:1.85-bullseye AS build
 
 RUN apt-get update
 RUN apt-get install protobuf-compiler -y

examples/flatmap-stream/Cargo.toml

@@ -1,9 +1,10 @@
 [package]
 name = "flatmap-stream"
 version = "0.1.0"
-edition = "2021"
+edition.workspace = true
+rust-version.workspace = true
 
 [dependencies]
-tonic = "0.13.0"
-tokio = { version = "1.0", features = ["macros", "rt-multi-thread"] }
+tonic.workspace = true
+tokio.workspace = true
 numaflow = { path = "../../numaflow" }

examples/flatmap-stream/Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.84-bullseye AS build
+FROM rust:1.85-bullseye AS build
 
 RUN apt-get update
 RUN apt-get install protobuf-compiler -y

examples/map-cat/Cargo.toml

@@ -1,9 +1,10 @@
 [package]
 name = "map-cat"
 version = "0.1.0"
-edition = "2021"
+edition.workspace = true
+rust-version.workspace = true
 
 [dependencies]
-tonic = "0.13.0"
-tokio = { version = "1.0", features = ["macros", "rt-multi-thread"] }
+tonic.workspace = true
+tokio.workspace = true
 numaflow = { path = "../../numaflow" }

examples/map-cat/Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.84-bullseye AS build
+FROM rust:1.85-bullseye AS build
 
 RUN apt-get update
 RUN apt-get install protobuf-compiler -y

examples/map-cat/src/main.rs

@@ -1,8 +1,12 @@
 use numaflow::map;
+use numaflow::shared::grpc_server::ServerExtras;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    map::Server::new(Cat).start().await
+    map::Server::new(Cat)
+        .with_max_message_size(10240)
+        .start()
+        .await
 }
 
 struct Cat;

examples/map-even-odd/.dockerignore

@@ -0,0 +1 @@
+target/

examples/map-even-odd/Cargo.toml

@@ -0,0 +1,10 @@
+[package]
+name = "map-even-odd"
+version = "0.1.0"
+edition.workspace = true
+rust-version.workspace = true
+
+[dependencies]
+tonic.workspace = true
+tokio.workspace = true
+numaflow = { path = "../../numaflow" }

examples/map-even-odd/Dockerfile

@@ -0,0 +1,20 @@
+FROM rust:1.85-bullseye AS build
+
+RUN apt-get update
+RUN apt-get install protobuf-compiler -y
+
+WORKDIR /numaflow-rs
+COPY ./ ./
+WORKDIR /numaflow-rs/examples/map-even-odd
+
+# build for release
+RUN cargo build --release
+
+# our final base
+FROM debian:bullseye AS map-even-odd
+
+# copy the build artifact from the build stage
+COPY --from=build /numaflow-rs/target/release/map-even-odd .
+
+# set the startup command to run your binary
+CMD ["./map-even-odd"]