Merge remote-tracking branch 'fs/main' into streaming-storage

# Conflicts:
#	Makefile
#	src/runtime/streaming/operators/grouping/incremental_aggregate.rs

Author: luoluoyuyu

.github/workflows/integration-test.yml

@@ -0,0 +1,76 @@
+# Licensed 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.
+
+name: Integration Tests
+
+on:
+  push:
+    branches: ["main", "master"]
+  pull_request:
+    branches: ["main", "master"]
+  workflow_dispatch:
+
+env:
+  CARGO_TERM_COLOR: always
+
+jobs:
+  integration-test:
+    name: Integration Tests
+    runs-on: ubuntu-latest
+    timeout-minutes: 180
+
+    steps:
+      - name: Checkout Source
+        uses: actions/checkout@v4
+
+      - name: Verify Docker
+        run: docker version
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: "pip"
+
+      - name: Setup Rust
+        uses: dtolnay/rust-toolchain@stable
+
+      - name: Install System Dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y --no-install-recommends \
+            cmake \
+            libssl-dev \
+            libcurl4-openssl-dev \
+            pkg-config \
+            libsasl2-dev \
+            protobuf-compiler
+
+      - name: Cache Cargo
+        uses: Swatinem/rust-cache@v2
+
+      - name: Build Release Binary
+        run: make env && make dist
+
+      - name: Pre-pull Kafka Image
+        run: docker pull apache/kafka:3.7.0
+
+      - name: Run Integration Tests
+        run: make integration-test
+
+      - name: Upload Test Logs
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: integration-test-logs
+          path: tests/integration/target/**/logs/
+          retention-days: 30

.gitignore

@@ -33,3 +33,8 @@ python/**/target/
 **/**.egg-info
 .cache/
 /.cursor/worktrees.json
+
+# Integration test output
+tests/integration/target/
+tests/integration/.venv/
+tests/integration/install

Makefile

@@ -79,7 +79,7 @@ C_0 := \033[0m
 log = @printf "$(C_B)[-]$(C_0) %-15s %s\n" "$(1)" "$(2)"
 success = @printf "$(C_G)[✔]$(C_0) %s\n" "$(1)"
 
-.PHONY: all help build build-lite dist dist-lite clean test env env-clean go-sdk-env go-sdk-build go-sdk-clean docker docker-run docker-push .check-env .ensure-target .build-wasm
+.PHONY: all help build build-lite dist dist-lite clean test env env-clean go-sdk-env go-sdk-build go-sdk-clean docker docker-run docker-push .check-env .build-wasm integration-test
 
 all: build
 
@@ -100,6 +100,8 @@ help:
 	@echo "  docker-run  Run container (port 8080, mount logs)"
 	@echo "  docker-push Push image to registry"
 	@echo ""
+	@echo "  integration-test   Run integration tests (delegates to tests/integration)"
+	@echo ""
 	@echo "  Version: $(VERSION) | Arch: $(ARCH) | OS: $(OS)"
 
 # 4. Auto-install missing Rust target toolchain
@@ -206,6 +208,7 @@ clean:
 	@cargo clean
 	@rm -rf $(DIST_ROOT) data logs
 	@./scripts/clean.sh 2>/dev/null || true
+	@$(MAKE) -C tests/integration clean 2>/dev/null || true
 	$(call success,Done)
 
 .check-env:
@@ -228,3 +231,6 @@ docker-push:
 	$(call log,DOCKER,Pushing $(IMAGE_NAME))
 	@docker push $(IMAGE_NAME)
 	$(call success,Push Complete)
+
+integration-test:
+	@$(MAKE) -C tests/integration test PYTEST_ARGS="$(PYTEST_ARGS)"

src/runtime/wasm/processor/wasm/wasm_task.rs

@@ -392,10 +392,27 @@ impl WasmTask {
     ) -> ControlAction {
         match signal {
             TaskControlSignal::Start { completion_flag } => {
-                for input in inputs.iter_mut() {
-                    let _ = input.start();
+                for (idx, input) in inputs.iter_mut().enumerate() {
+                    if let Err(e) = input.start() {
+                        let msg = format!("Failed to start input {}: {}", idx, e);
+                        log::error!("{}", msg);
+                        *failure_cause.lock().unwrap() = Some(msg.clone());
+                        *shared_state.lock().unwrap() =
+                            ComponentState::Error { error: msg.clone() };
+                        *execution_state.lock().unwrap() = ExecutionState::Failed;
+                        completion_flag.mark_error(msg);
+                        return ControlAction::Pause;
+                    }
+                }
+                if let Err(e) = processor.start_outputs() {
+                    let msg = format!("Failed to start outputs: {}", e);
+                    log::error!("{}", msg);
+                    *failure_cause.lock().unwrap() = Some(msg.clone());
+                    *shared_state.lock().unwrap() = ComponentState::Error { error: msg.clone() };
+                    *execution_state.lock().unwrap() = ExecutionState::Failed;
+                    completion_flag.mark_error(msg);
+                    return ControlAction::Pause;
                 }
-                let _ = processor.start_outputs();
                 *state = TaskState::Running;
                 *shared_state.lock().unwrap() = ComponentState::Running;
                 completion_flag.mark_completed();

src/server/handler.rs

@@ -91,6 +91,25 @@ impl FunctionStreamServiceImpl {
         }
     }
 
+    fn classify_error(message: &str) -> StatusCode {
+        let lower = message.to_lowercase();
+        if lower.contains("not found") || lower.contains("not exist") {
+            StatusCode::NotFound
+        } else if lower.contains("uniqueness violation")
+            || lower.contains("already exists")
+            || lower.contains("duplicate")
+        {
+            StatusCode::Conflict
+        } else if lower.contains("invalid")
+            || lower.contains("unsupported")
+            || lower.contains("missing")
+        {
+            StatusCode::BadRequest
+        } else {
+            StatusCode::InternalServerError
+        }
+    }
+
     async fn execute_statement(
         &self,
         stmt: &dyn Statement,
@@ -101,7 +120,8 @@ impl FunctionStreamServiceImpl {
         if result.success {
             Self::build_success_response(success_status, result.message, result.data)
         } else {
-            Self::build_error_response(StatusCode::InternalServerError, result.message)
+            let status = Self::classify_error(&result.message);
+            Self::build_error_response(status, result.message)
         }
     }
 }
@@ -139,8 +159,9 @@ impl FunctionStreamService for FunctionStreamServiceImpl {
 
             if !result.success {
                 error!("SQL execution aborted: {}", result.message);
+                let status = Self::classify_error(&result.message);
                 return Ok(TonicResponse::new(Self::build_error_response(
-                    StatusCode::InternalServerError,
+                    status,
                     result.message,
                 )));
             }
@@ -235,8 +256,9 @@ impl FunctionStreamService for FunctionStreamServiceImpl {
 
         if !result.success {
             error!("show_functions execution failed: {}", result.message);
+            let status = Self::classify_error(&result.message);
             return Ok(TonicResponse::new(ShowFunctionsResponse {
-                status_code: StatusCode::InternalServerError as i32,
+                status_code: status as i32,
                 message: result.message,
                 functions: vec![],
             }));

tests/integration/Makefile

@@ -0,0 +1,64 @@
+# Licensed 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.
+
+# -----------------------------------------------------------------------
+# Integration Test Makefile
+# -----------------------------------------------------------------------
+# Usage:
+#   make test        — Setup env + run pytest  (PYTEST_ARGS="-k xxx")
+#   make clean       — Remove .venv and test output
+#
+# Prerequisites:
+#   The FunctionStream binary must already be built (make build / make build-lite
+#   from the project root).
+# -----------------------------------------------------------------------
+
+PROJECT_ROOT := $(shell git -C $(CURDIR) rev-parse --show-toplevel)
+PYTHON_ROOT  := $(PROJECT_ROOT)/python
+VENV         := $(CURDIR)/.venv
+PIP          := $(VENV)/bin/pip
+PY           := $(VENV)/bin/python
+
+C_G := \033[0;32m
+C_B := \033[0;34m
+C_0 := \033[0m
+
+log     = @printf "$(C_B)[-]$(C_0) %-12s %s\n" "$(1)" "$(2)"
+success = @printf "$(C_G)[✔]$(C_0) %s\n" "$(1)"
+
+.PHONY: test clean help
+
+help:
+	@echo "Integration Test Targets:"
+	@echo ""
+	@echo "  test    Setup Python env + run pytest (PYTEST_ARGS=...)"
+	@echo "  clean   Remove .venv and target/tests output"
+
+install: requirements.txt $(PYTHON_ROOT)/functionstream-api/pyproject.toml $(PYTHON_ROOT)/functionstream-client/pyproject.toml
+	$(call log,ENV,Setting up Python virtual environment)
+	@test -d $(VENV) || python3 -m venv $(VENV)
+	@$(PIP) install --quiet --upgrade pip
+	@$(PIP) install --quiet -r requirements.txt
+	@touch $@
+	$(call success,Python environment ready)
+
+test: install
+	$(call log,TEST,Running integration tests)
+	@$(PY) -m pytest -v $(PYTEST_ARGS)
+	$(call success,All integration tests passed)
+
+clean:
+	$(call log,CLEAN,Removing test artifacts)
+	@rm -rf $(VENV)
+	@rm -rf $(CURDIR)/target
+	@rm -rf $(CURDIR)/install
+	$(call success,Clean complete)

tests/integration/README.md

@@ -0,0 +1,88 @@
+# Integration Tests
+
+## Prerequisites
+
+| Dependency | Version  | Purpose                                         |
+|------------|----------|-------------------------------------------------|
+| Python     | >= 3.9   | Test framework runtime                          |
+| Rust       | stable   | Build the FunctionStream binary                 |
+| Docker     | >= 20.10 | Run a Kafka broker for streaming integration tests |
+
+> **Docker is required.** The test framework automatically pulls and manages
+> an `apache/kafka:3.7.0` container in KRaft mode to provide a real Kafka
+> broker for tests that involve Kafka input/output. Tests will fail if the
+> Docker daemon is not running.
+
+## Quick Start
+
+```bash
+# From the project root
+make build          # Build the release binary (with --features python)
+make integration-test
+
+# Or run directly from this directory
+cd tests/integration
+make test
+```
+
+## Directory Layout
+
+```
+tests/integration/
+├── Makefile              # test / clean targets
+├── requirements.txt      # Python dependencies (pytest, grpcio, docker, ...)
+├── pytest.ini            # Pytest configuration
+├── framework/            # Reusable test infrastructure
+│   ├── instance.py       # FunctionStreamInstance facade
+│   ├── workspace.py      # Per-test directory management
+│   ├── config.py         # Server config generation
+│   ├── process.py        # OS process lifecycle (start/stop/kill)
+│   ├── utils.py          # Port allocation, readiness probes
+│   └── kafka_manager.py  # Docker-managed Kafka broker (KRaft mode)
+├── test/                 # Test suites
+│   ├── wasm/             # WASM function tests
+│   │   └── python_sdk/   # Python SDK integration tests
+│   └── streaming/        # Streaming engine tests (future)
+└── target/               # Test output (git-ignored)
+    ├── .shared_cache/    # Shared WASM compilation cache across tests
+    └── <suite>/<class>/<method>/<timestamp>/logs/
+```
+
+## Test Output
+
+Each test gets an isolated server instance with its own log directory:
+
+```
+target/wasm/python_sdk/TestFunctionLifecycle/test_full_lifecycle_transitions/20260416_221655/
+  logs/
+    app.log       # FunctionStream application log
+    stdout.log    # Server stdout
+    stderr.log    # Server stderr
+```
+
+Only `logs/` is retained after tests complete; `conf/` and `data/` are
+automatically cleaned up.
+
+## Python Dependencies
+
+All Python packages are listed in `requirements.txt` and installed
+automatically by `make test`. Key dependencies:
+
+- `pytest` — test runner
+- `grpcio` / `protobuf` — gRPC client communication
+- `docker` — Docker SDK for managing the Kafka container
+- `confluent-kafka` — Kafka admin client for topic management
+- `functionstream-api` / `functionstream-client` — local editable installs
+
+## Running Specific Tests
+
+```bash
+# Single test
+make test PYTEST_ARGS="-k test_full_lifecycle_transitions"
+
+# Single file
+make test PYTEST_ARGS="test/wasm/python_sdk/test_lifecycle.py"
+
+# Verbose with live log
+make test PYTEST_ARGS="-v --log-cli-level=DEBUG"
+```

tests/integration/framework/__init__.py

@@ -0,0 +1,22 @@
+# Licensed 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.
+
+from .instance import FunctionStreamInstance
+
+__all__ = ["FunctionStreamInstance", "KafkaDockerManager"]
+
+
+def __getattr__(name: str):
+    if name == "KafkaDockerManager":
+        from .kafka_manager import KafkaDockerManager
+        return KafkaDockerManager
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")