Merge branch 'main' into kwannoel/bump-unit-test-timeout

ci/workflows/main-cron.yml

@@ -480,3 +480,104 @@ steps:
       - ./ci/plugins/upload-failure-logs
     timeout_in_minutes: 21
     retry: *auto-retry
+
+  - label: "end-to-end test for opendal (parallel)"
+    command: "ci/scripts/e2e-test-parallel-for-opendal.sh -p ci-release"
+    depends_on:
+      - "build"
+      - "docslt"
+    plugins:
+      - docker-compose#v4.9.0:
+          run: rw-build-env
+          config: ci/docker-compose.yml
+          mount-buildkite-agent: true
+      - ./ci/plugins/upload-failure-logs
+    timeout_in_minutes: 14
+    retry: *auto-retry
+
+  - label: "end-to-end test (parallel, in-memory)"
+    command: "ci/scripts/e2e-test-parallel-in-memory.sh -p ci-release"
+    depends_on: "build"
+    plugins:
+      - docker-compose#v4.9.0:
+          run: rw-build-env
+          config: ci/docker-compose.yml
+          mount-buildkite-agent: true
+      - ./ci/plugins/upload-failure-logs
+    timeout_in_minutes: 12
+    retry: *auto-retry
+
+  - label: "end-to-end iceberg sink test"
+    command: "ci/scripts/e2e-iceberg-sink-test.sh -p ci-release"
+    depends_on:
+      - "build"
+      - "build-other"
+    plugins:
+      - docker-compose#v4.9.0:
+          run: rw-build-env
+          config: ci/docker-compose.yml
+          mount-buildkite-agent: true
+      - ./ci/plugins/upload-failure-logs
+    timeout_in_minutes: 5
+    retry: *auto-retry
+
+  - label: "end-to-end iceberg sink v2 test"
+    command: "ci/scripts/e2e-iceberg-sink-v2-test.sh -p ci-release"
+    depends_on:
+      - "build"
+      - "build-other"
+    plugins:
+      - docker-compose#v4.9.0:
+          run: sink-test-env
+          config: ci/docker-compose.yml
+          mount-buildkite-agent: true
+      - ./ci/plugins/upload-failure-logs
+    timeout_in_minutes: 10
+    retry: *auto-retry
+
+  - label: "end-to-end clickhouse sink test"
+    command: "ci/scripts/e2e-clickhouse-sink-test.sh -p ci-release"
+    depends_on:
+      - "build"
+      - "build-other"
+    plugins:
+      - docker-compose#v4.9.0:
+          run: sink-test-env
+          config: ci/docker-compose.yml
+          mount-buildkite-agent: true
+      - ./ci/plugins/upload-failure-logs
+    timeout_in_minutes: 10
+    retry: *auto-retry
+
+  - label: "e2e java-binding test"
+    command: "ci/scripts/java-binding-test.sh -p ci-release"
+    depends_on:
+      - "build"
+      - "build-other"
+    plugins:
+      - docker-compose#v4.9.0:
+          run: rw-build-env
+          config: ci/docker-compose.yml
+          mount-buildkite-agent: true
+      - ./ci/plugins/upload-failure-logs
+    timeout_in_minutes: 10
+    retry: *auto-retry
+
+  - label: "connector node integration test Java {{matrix.java_version}}"
+    command: "ci/scripts/connector-node-integration-test.sh -p ci-release -v {{matrix.java_version}}"
+    depends_on:
+      - "build"
+      - "build-other"
+    plugins:
+      - docker-compose#v4.9.0:
+          run: rw-build-env
+          config: ci/docker-compose.yml
+          mount-buildkite-agent: true
+      - ./ci/plugins/upload-failure-logs
+    matrix:
+      setup:
+        java_version:
+          - "11"
+          - "17"
+    timeout_in_minutes: 10
+    retry: *auto-retry

ci/workflows/pull-request.yml

@@ -95,6 +95,7 @@ steps:
     retry: *auto-retry
 
   - label: "end-to-end test for opendal (parallel)"
+    if: build.pull_request.labels includes "ci/run-opendal-tests"
     command: "ci/scripts/e2e-test-parallel-for-opendal.sh -p ci-dev"
     depends_on:
       - "build"
@@ -109,6 +110,7 @@ steps:
     retry: *auto-retry
 
   - label: "end-to-end test (parallel, in-memory)"
+    if: build.pull_request.labels includes "ci/run-e2e-parallel-in-memory-tests"
     command: "ci/scripts/e2e-test-parallel-in-memory.sh -p ci-dev"
     depends_on: "build"
     plugins:
@@ -150,6 +152,7 @@ steps:
     retry: *auto-retry
 
   - label: "connector node integration test Java {{matrix.java_version}}"
+    if: build.pull_request.labels includes "ci/run-java-connector-node-integration-tests"
     command: "ci/scripts/connector-node-integration-test.sh -p ci-dev -v {{matrix.java_version}}"
     depends_on:
       - "build"
@@ -169,6 +172,7 @@ steps:
     retry: *auto-retry
 
   - label: "end-to-end iceberg sink test"
+    if: build.pull_request.labels includes "ci/run-e2e-iceberg-sink-tests"
     command: "ci/scripts/e2e-iceberg-sink-test.sh -p ci-dev"
     depends_on:
       - "build"
@@ -183,6 +187,7 @@ steps:
     retry: *auto-retry
 
   - label: "end-to-end iceberg sink v2 test"
+    if: build.pull_request.labels includes "ci/run-e2e-iceberg-sink-tests"
     command: "ci/scripts/e2e-iceberg-sink-v2-test.sh -p ci-dev"
     depends_on:
       - "build"
@@ -197,6 +202,7 @@ steps:
     retry: *auto-retry
 
   - label: "end-to-end clickhouse sink test"
+    if: build.pull_request.labels includes "ci/run-e2e-clickhouse-sink-tests"
     command: "ci/scripts/e2e-clickhouse-sink-test.sh -p ci-dev"
     depends_on:
       - "build"
@@ -211,6 +217,7 @@ steps:
     retry: *auto-retry
 
   - label: "e2e java-binding test"
+    if: build.pull_request.labels includes "ci/run-java-binding-tests"
     command: "ci/scripts/java-binding-test.sh -p ci-dev"
     depends_on:
       - "build"

proto/meta.proto

@@ -495,7 +495,7 @@ service MetaMemberService {
 
 // The schema for persisted system parameters.
 // Note on backward compatibility:
-// - Do not remove deprecated fields. Mark them as deprecated both after the field definition and in `system_params/mod.rs` instead.
+// - Do not remove deprecated fields. Mark them as deprecated instead.
 // - Do not rename existing fields, since each field is stored separately in the meta store with the field name as the key.
 // - To modify (rename, change the type or semantic of) a field, introduce a new field suffixed by the version.
 message SystemParams {

src/common/src/system_param/mod.rs

@@ -33,16 +33,15 @@ pub type SystemParamsError = String;
 
 type Result<T> = core::result::Result<T, SystemParamsError>;
 
-/// Only includes undeprecated params.
+/// Define all system parameters here.
+///
 /// Macro input is { field identifier, type, default value, is mutable }
 ///
 /// Note:
 /// - Having `None` as default value means the parameter must be initialized.
 #[macro_export]
-macro_rules! for_all_undeprecated_params {
-    ($macro:ident
-        // Hack: match trailing fields to implement `for_all_params`
-        $(, { $field:ident, $type:ty, $default:expr, $is_mutable:expr })*) => {
+macro_rules! for_all_params {
+    ($macro:ident) => {
         $macro! {
             { barrier_interval_ms, u32, Some(1000_u32), true },
             { checkpoint_frequency, u64, Some(1_u64), true },
@@ -56,22 +55,10 @@ macro_rules! for_all_undeprecated_params {
             { backup_storage_directory, String, Some("backup".to_string()), false },
             { max_concurrent_creating_streaming_jobs, u32, Some(1_u32), true },
             { pause_on_next_bootstrap, bool, Some(false), true },
-            $({ $field, $type, $default, $is_mutable },)*
         }
     };
 }
 
-/// Includes all params.
-/// Macro input is same as `for_all_undeprecated_params`.
-macro_rules! for_all_params {
-    ($macro:ident) => {
-        for_all_undeprecated_params!(
-            $macro /* Define future deprecated params here, such as
-                    * ,{ backup_storage_directory, String, "backup".to_string(), true } */
-        );
-    };
-}
-
 /// Convert field name to string.
 #[macro_export]
 macro_rules! key_of {
@@ -106,7 +93,7 @@ macro_rules! def_default {
     };
 }
 
-for_all_undeprecated_params!(def_default);
+for_all_params!(def_default);
 
 macro_rules! impl_check_missing_fields {
     ($({ $field:ident, $type:ty, $default:expr, $is_mutable:expr },)*) => {
@@ -186,10 +173,9 @@ macro_rules! impl_system_params_from_kv {
                         std::str::from_utf8(v.as_ref()).unwrap().to_string()
                     )
                 }).collect::<Vec<_>>();
-                Err(format!("unrecognized system params {:?}", unrecognized_params))
-            } else {
-                Ok(ret)
+                tracing::warn!("unrecognized system params {:?}", unrecognized_params);
             }
+            Ok(ret)
         }
     };
 }
@@ -317,12 +303,12 @@ macro_rules! impl_system_params_for_test {
 
 for_all_params!(impl_system_params_from_kv);
 for_all_params!(impl_is_mutable);
-for_all_undeprecated_params!(impl_derive_missing_fields);
-for_all_undeprecated_params!(impl_check_missing_fields);
-for_all_undeprecated_params!(impl_system_params_to_kv);
-for_all_undeprecated_params!(impl_set_system_param);
-for_all_undeprecated_params!(impl_default_validation_on_set);
-for_all_undeprecated_params!(impl_system_params_for_test);
+for_all_params!(impl_derive_missing_fields);
+for_all_params!(impl_check_missing_fields);
+for_all_params!(impl_system_params_to_kv);
+for_all_params!(impl_set_system_param);
+for_all_params!(impl_default_validation_on_set);
+for_all_params!(impl_system_params_for_test);
 
 struct OverrideValidateOnSet;
 impl ValidateOnSet for OverrideValidateOnSet {
@@ -345,7 +331,7 @@ impl ValidateOnSet for OverrideValidateOnSet {
     }
 }
 
-for_all_undeprecated_params!(impl_default_from_other_params);
+for_all_params!(impl_default_from_other_params);
 
 struct OverrideFromParams;
 impl FromParams for OverrideFromParams {}
@@ -370,14 +356,15 @@ mod tests {
             (BACKUP_STORAGE_DIRECTORY_KEY, "a"),
             (MAX_CONCURRENT_CREATING_STREAMING_JOBS_KEY, "1"),
             (PAUSE_ON_NEXT_BOOTSTRAP_KEY, "false"),
+            ("a_deprecated_param", "foo"),
         ];
 
         // To kv - missing field.
         let p = PbSystemParams::default();
         assert!(system_params_to_kv(&p).is_err());
 
-        // From kv - unrecognized field.
-        assert!(system_params_from_kv(vec![("?", "?")]).is_err());
+        // From kv - unrecognized field should be ignored
+        assert!(system_params_from_kv(vec![("?", "?")]).is_ok());
 
         // Deser & ser.
         let p = system_params_from_kv(kvs).unwrap();

src/meta/src/manager/system_param/mod.rs

@@ -21,7 +21,7 @@ use std::time::Duration;
 use anyhow::anyhow;
 use risingwave_common::system_param::reader::SystemParamsReader;
 use risingwave_common::system_param::{check_missing_params, set_system_param};
-use risingwave_common::{for_all_undeprecated_params, key_of};
+use risingwave_common::{for_all_params, key_of};
 use risingwave_pb::meta::subscribe_response::{Info, Operation};
 use risingwave_pb::meta::SystemParams;
 use tokio::sync::oneshot::Sender;
@@ -182,4 +182,4 @@ macro_rules! impl_merge_params {
     };
 }
 
-for_all_undeprecated_params!(impl_merge_params);
+for_all_params!(impl_merge_params);