risingwave_meta_node/
lib.rs

1// Copyright 2023 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15#![cfg_attr(coverage, feature(coverage_attribute))]
16
17mod server;
18
19use std::path::PathBuf;
20use std::time::Duration;
21
22use clap::Parser;
23use educe::Educe;
24pub use error::{MetaError, MetaResult};
25use redact::Secret;
26use risingwave_common::config::OverrideConfig;
27use risingwave_common::license::LicenseKey;
28use risingwave_common::system_param::StateStoreUrl;
29use risingwave_common::util::meta_addr::MetaAddressStrategy;
30use risingwave_common::util::resource_util;
31use risingwave_common::util::tokio_util::sync::CancellationToken;
32use risingwave_common::{GIT_SHA, RW_VERSION};
33use risingwave_common_heap_profiling::HeapProfiler;
34use risingwave_meta::*;
35use risingwave_meta_service::*;
36pub use rpc::{ElectionClient, ElectionMember};
37use server::rpc_serve;
38pub use server::started::get as is_server_started;
39
40use crate::manager::MetaOpts;
41
42#[derive(Educe, Clone, Parser, OverrideConfig)]
43#[educe(Debug)]
44#[command(version, about = "The central metadata management service")]
45pub struct MetaNodeOpts {
46    // TODO: use `SocketAddr`
47    #[clap(long, env = "RW_LISTEN_ADDR", default_value = "127.0.0.1:5690")]
48    pub listen_addr: String,
49
50    /// The address for contacting this instance of the service.
51    /// This would be synonymous with the service's "public address"
52    /// or "identifying address".
53    /// It will serve as a unique identifier in cluster
54    /// membership and leader election. Must be specified for meta backend.
55    #[clap(long, env = "RW_ADVERTISE_ADDR", default_value = "127.0.0.1:5690")]
56    pub advertise_addr: String,
57
58    #[clap(long, env = "RW_DASHBOARD_HOST")]
59    pub dashboard_host: Option<String>,
60
61    /// We will start a http server at this address via `MetricsManager`.
62    /// Then the prometheus instance will poll the metrics from this address.
63    #[clap(long, env = "RW_PROMETHEUS_HOST", alias = "prometheus-host")]
64    pub prometheus_listener_addr: Option<String>,
65
66    /// Endpoint of the SQL service, make it non-option when SQL service is required.
67    #[clap(long, hide = true, env = "RW_SQL_ENDPOINT")]
68    pub sql_endpoint: Option<Secret<String>>,
69
70    /// Username of sql backend, required when meta backend set to MySQL or PostgreSQL.
71    #[clap(long, hide = true, env = "RW_SQL_USERNAME", default_value = "")]
72    pub sql_username: String,
73
74    /// Password of sql backend, required when meta backend set to MySQL or PostgreSQL.
75    #[clap(long, hide = true, env = "RW_SQL_PASSWORD", default_value = "")]
76    pub sql_password: Secret<String>,
77
78    /// Database of sql backend, required when meta backend set to MySQL or PostgreSQL.
79    #[clap(long, hide = true, env = "RW_SQL_DATABASE", default_value = "")]
80    pub sql_database: String,
81
82    /// Params for the URL connection, such as `sslmode=disable`.
83    /// Example: `param1=value1&param2=value2`
84    #[clap(long, hide = true, env = "RW_SQL_URL_PARAMS")]
85    pub sql_url_params: Option<String>,
86
87    /// The HTTP REST-API address of the Prometheus instance associated to this cluster.
88    /// This address is used to serve `PromQL` queries to Prometheus.
89    /// It is also used by Grafana Dashboard Service to fetch metrics and visualize them.
90    #[clap(long, env = "RW_PROMETHEUS_ENDPOINT")]
91    pub prometheus_endpoint: Option<String>,
92
93    /// The additional selector used when querying Prometheus.
94    ///
95    /// The format is same as `PromQL`. Example: `instance="foo",namespace="bar"`
96    #[clap(long, env = "RW_PROMETHEUS_SELECTOR")]
97    pub prometheus_selector: Option<String>,
98
99    /// Default tag for the endpoint created when creating a privatelink connection.
100    /// Will be appended to the tags specified in the `tags` field in with clause in `create
101    /// connection`.
102    #[clap(long, hide = true, env = "RW_PRIVATELINK_ENDPOINT_DEFAULT_TAGS")]
103    pub privatelink_endpoint_default_tags: Option<String>,
104
105    #[clap(long, hide = true, env = "RW_VPC_ID")]
106    pub vpc_id: Option<String>,
107
108    #[clap(long, hide = true, env = "RW_VPC_SECURITY_GROUP_ID")]
109    pub security_group_id: Option<String>,
110
111    /// The path of `risingwave.toml` configuration file.
112    ///
113    /// If empty, default configuration values will be used.
114    #[clap(long, env = "RW_CONFIG_PATH", default_value = "")]
115    pub config_path: String,
116
117    #[clap(long, hide = true, env = "RW_BACKEND", value_enum)]
118    #[override_opts(path = meta.backend)]
119    pub backend: Option<MetaBackend>,
120
121    /// The interval of periodic barrier.
122    #[clap(long, hide = true, env = "RW_BARRIER_INTERVAL_MS")]
123    #[override_opts(path = system.barrier_interval_ms)]
124    pub barrier_interval_ms: Option<u32>,
125
126    /// Target size of the Sstable.
127    #[clap(long, hide = true, env = "RW_SSTABLE_SIZE_MB")]
128    #[override_opts(path = system.sstable_size_mb)]
129    pub sstable_size_mb: Option<u32>,
130
131    /// Size of each block in bytes in SST.
132    #[clap(long, hide = true, env = "RW_BLOCK_SIZE_KB")]
133    #[override_opts(path = system.block_size_kb)]
134    pub block_size_kb: Option<u32>,
135
136    /// Deprecated: Bloom filter is no longer a supported SST filter implementation.
137    #[clap(long, hide = true, env = "RW_BLOOM_FALSE_POSITIVE")]
138    #[override_opts(path = system.bloom_false_positive)]
139    pub bloom_false_positive: Option<f64>,
140
141    /// State store url
142    #[clap(long, hide = true, env = "RW_STATE_STORE")]
143    #[override_opts(path = system.state_store)]
144    pub state_store: Option<StateStoreUrl>,
145
146    /// Remote directory for storing data and metadata objects.
147    #[clap(long, hide = true, env = "RW_DATA_DIRECTORY")]
148    #[override_opts(path = system.data_directory)]
149    pub data_directory: Option<String>,
150
151    /// Whether config object storage bucket lifecycle to purge stale data.
152    #[clap(long, hide = true, env = "RW_DO_NOT_CONFIG_BUCKET_LIFECYCLE")]
153    #[override_opts(path = meta.do_not_config_object_storage_lifecycle)]
154    pub do_not_config_object_storage_lifecycle: Option<bool>,
155
156    /// Remote storage url for storing snapshots.
157    #[clap(long, hide = true, env = "RW_BACKUP_STORAGE_URL")]
158    #[override_opts(path = system.backup_storage_url)]
159    pub backup_storage_url: Option<String>,
160
161    /// Remote directory for storing snapshots.
162    #[clap(long, hide = true, env = "RW_BACKUP_STORAGE_DIRECTORY")]
163    #[override_opts(path = system.backup_storage_directory)]
164    pub backup_storage_directory: Option<String>,
165
166    /// Enable heap profile dump when memory usage is high.
167    #[clap(long, hide = true, env = "RW_HEAP_PROFILING_DIR")]
168    #[override_opts(path = server.heap_profiling.dir)]
169    pub heap_profiling_dir: Option<String>,
170
171    /// Exit if idle for a certain period of time.
172    #[clap(long, hide = true, env = "RW_DANGEROUS_MAX_IDLE_SECS")]
173    #[override_opts(path = meta.dangerous_max_idle_secs)]
174    pub dangerous_max_idle_secs: Option<u64>,
175
176    /// Endpoint of the connector node.
177    #[deprecated = "connector node has been deprecated."]
178    #[clap(long, hide = true, env = "RW_CONNECTOR_RPC_ENDPOINT")]
179    pub connector_rpc_endpoint: Option<String>,
180
181    /// The license key to activate enterprise features.
182    #[clap(long, hide = true, env = "RW_LICENSE_KEY")]
183    #[override_opts(path = system.license_key)]
184    pub license_key: Option<LicenseKey>,
185
186    /// The path of the license key file to be watched and hot-reloaded.
187    #[clap(long, env = "RW_LICENSE_KEY_PATH")]
188    pub license_key_path: Option<PathBuf>,
189
190    /// 128-bit AES key for secret store in HEX format.
191    #[clap(long, hide = true, env = "RW_SECRET_STORE_PRIVATE_KEY_HEX")]
192    pub secret_store_private_key_hex: Option<Secret<String>>,
193
194    /// The path of the temp secret file directory.
195    #[clap(
196        long,
197        hide = true,
198        env = "RW_TEMP_SECRET_FILE_DIR",
199        default_value = "./secrets"
200    )]
201    pub temp_secret_file_dir: String,
202
203    /// Address of the serverless backfill controller.
204    /// Needed if meta receives a streaming job with serverless backfill enabled.
205    /// Feature disabled by default.
206    #[clap(long, env = "RW_SBC_ADDR", default_value = "")]
207    pub serverless_backfill_controller_addr: String,
208}
209
210impl risingwave_common::opts::Opts for MetaNodeOpts {
211    fn name() -> &'static str {
212        "meta"
213    }
214
215    fn meta_addr(&self) -> MetaAddressStrategy {
216        format!("http://{}", self.listen_addr)
217            .parse()
218            .expect("invalid listen address")
219    }
220}
221
222use std::future::Future;
223use std::pin::Pin;
224use std::sync::Arc;
225
226use risingwave_common::config::{MetaBackend, RwConfig, load_config};
227use tracing::info;
228
229/// Start meta node
230pub fn start(
231    opts: MetaNodeOpts,
232    shutdown: CancellationToken,
233) -> Pin<Box<dyn Future<Output = ()> + Send>> {
234    // WARNING: don't change the function signature. Making it `async fn` will cause
235    // slow compile in release mode.
236    Box::pin(async move {
237        info!("Starting meta node");
238        info!("> options: {:?}", opts);
239        let config = load_config(&opts.config_path, &opts);
240        info!("> config: {:?}", config);
241        info!("> version: {} ({})", RW_VERSION, GIT_SHA);
242        let listen_addr = opts.listen_addr.parse().unwrap();
243        let dashboard_addr = opts.dashboard_host.map(|x| x.parse().unwrap());
244        let prometheus_addr = opts.prometheus_listener_addr.map(|x| x.parse().unwrap());
245        let meta_store_config = config.meta.meta_store_config.clone();
246        let backend = match config.meta.backend {
247            MetaBackend::Mem => {
248                if opts.sql_endpoint.is_some() {
249                    tracing::warn!("`--sql-endpoint` is ignored when using `mem` backend");
250                }
251                MetaStoreBackend::Mem
252            }
253            MetaBackend::Sql => MetaStoreBackend::Sql {
254                endpoint: opts
255                    .sql_endpoint
256                    .expect("sql endpoint is required")
257                    .expose_secret()
258                    .clone(),
259                config: meta_store_config,
260            },
261            MetaBackend::Sqlite => MetaStoreBackend::Sql {
262                endpoint: format!(
263                    "sqlite://{}?mode=rwc",
264                    opts.sql_endpoint
265                        .expect("sql endpoint is required")
266                        .expose_secret()
267                ),
268                config: meta_store_config,
269            },
270            MetaBackend::Postgres => MetaStoreBackend::Sql {
271                endpoint: format!(
272                    "postgres://{}:{}@{}/{}{}",
273                    opts.sql_username,
274                    opts.sql_password.expose_secret(),
275                    opts.sql_endpoint
276                        .expect("sql endpoint is required")
277                        .expose_secret(),
278                    opts.sql_database,
279                    if let Some(params) = &opts.sql_url_params
280                        && !params.is_empty()
281                    {
282                        format!("?{}", params)
283                    } else {
284                        "".to_owned()
285                    }
286                ),
287                config: meta_store_config,
288            },
289            MetaBackend::Mysql => MetaStoreBackend::Sql {
290                endpoint: format!(
291                    "mysql://{}:{}@{}/{}{}",
292                    opts.sql_username,
293                    opts.sql_password.expose_secret(),
294                    opts.sql_endpoint
295                        .expect("sql endpoint is required")
296                        .expose_secret(),
297                    opts.sql_database,
298                    if let Some(params) = &opts.sql_url_params
299                        && !params.is_empty()
300                    {
301                        format!("?{}", params)
302                    } else {
303                        "".to_owned()
304                    }
305                ),
306                config: meta_store_config,
307            },
308        };
309        validate_config(&config);
310
311        let total_memory_bytes = resource_util::memory::system_memory_available_bytes();
312        let heap_profiler =
313            HeapProfiler::new(total_memory_bytes, config.server.heap_profiling.clone());
314        // Run a background heap profiler
315        heap_profiler.start();
316
317        let secret_store_private_key = opts
318            .secret_store_private_key_hex
319            .map(|key| hex::decode(key.expose_secret()).unwrap());
320        let max_heartbeat_interval =
321            Duration::from_secs(config.meta.max_heartbeat_interval_secs as u64);
322        let max_idle_ms = config.meta.dangerous_max_idle_secs.unwrap_or(0) * 1000;
323        let in_flight_barrier_nums = config.streaming.in_flight_barrier_nums;
324        let snapshot_backfill_finish_max_lagged_barriers = config
325            .streaming
326            .snapshot_backfill_finish_max_lagged_barriers;
327        let snapshot_backfill_barrier_amplification_factor = config
328            .streaming
329            .snapshot_backfill_barrier_amplification_factor;
330        let privatelink_endpoint_default_tags =
331            opts.privatelink_endpoint_default_tags.map(|tags| {
332                tags.split(',')
333                    .map(|s| {
334                        let key_val = s.split_once('=').unwrap();
335                        (key_val.0.to_owned(), key_val.1.to_owned())
336                    })
337                    .collect()
338            });
339
340        let add_info = AddressInfo {
341            advertise_addr: opts.advertise_addr.clone(),
342            listen_addr,
343            prometheus_addr,
344            dashboard_addr,
345        };
346
347        const MIN_TIMEOUT_INTERVAL_SEC: u64 = 20;
348        let compaction_task_max_progress_interval_secs = {
349            let retry_config = &config.storage.object_store.retry;
350            let max_streaming_read_timeout_ms = (retry_config.streaming_read_attempt_timeout_ms
351                + retry_config.req_backoff_max_delay_ms)
352                * retry_config.streaming_read_retry_attempts as u64;
353            let max_streaming_upload_timeout_ms = (retry_config
354                .streaming_upload_attempt_timeout_ms
355                + retry_config.req_backoff_max_delay_ms)
356                * retry_config.streaming_upload_retry_attempts as u64;
357            let max_upload_timeout_ms = (retry_config.upload_attempt_timeout_ms
358                + retry_config.req_backoff_max_delay_ms)
359                * retry_config.upload_retry_attempts as u64;
360            let max_read_timeout_ms = (retry_config.read_attempt_timeout_ms
361                + retry_config.req_backoff_max_delay_ms)
362                * retry_config.read_retry_attempts as u64;
363            let max_timeout_ms = max_streaming_read_timeout_ms
364                .max(max_upload_timeout_ms)
365                .max(max_streaming_upload_timeout_ms)
366                .max(max_read_timeout_ms)
367                .max(config.meta.compaction_task_max_progress_interval_secs * 1000);
368            max_timeout_ms / 1000
369        } + MIN_TIMEOUT_INTERVAL_SEC;
370
371        Box::pin(rpc_serve(
372            add_info,
373            backend,
374            max_heartbeat_interval,
375            config.meta.meta_leader_lease_secs,
376            config.server.clone(),
377            MetaOpts {
378                enable_recovery: !config.meta.disable_recovery,
379                clean_all_foreground_jobs_on_recovery: config
380                    .meta
381                    .clean_all_foreground_jobs_on_recovery,
382                disable_automatic_parallelism_control: config
383                    .meta
384                    .disable_automatic_parallelism_control,
385                parallelism_control_batch_size: config.meta.parallelism_control_batch_size,
386                parallelism_control_trigger_period_sec: config
387                    .meta
388                    .parallelism_control_trigger_period_sec,
389                parallelism_control_trigger_first_delay_sec: config
390                    .meta
391                    .parallelism_control_trigger_first_delay_sec,
392                in_flight_barrier_nums,
393                snapshot_backfill_finish_max_lagged_barriers,
394                snapshot_backfill_barrier_amplification_factor,
395                max_idle_ms,
396                compaction_deterministic_test: config.meta.enable_compaction_deterministic,
397                default_parallelism: config.meta.default_parallelism,
398                vacuum_interval_sec: config.meta.vacuum_interval_sec,
399                time_travel_vacuum_interval_sec: config
400                    .meta
401                    .developer
402                    .time_travel_vacuum_interval_sec,
403                time_travel_vacuum_max_version_count: config
404                    .meta
405                    .developer
406                    .time_travel_vacuum_max_version_count,
407                vacuum_spin_interval_ms: config.meta.vacuum_spin_interval_ms,
408                iceberg_gc_interval_sec: config.meta.iceberg_gc_interval_sec,
409                iceberg_compaction_report_timeout_sec: config
410                    .meta
411                    .iceberg_compaction_report_timeout_sec,
412                iceberg_compaction_config_refresh_interval_sec: config
413                    .meta
414                    .iceberg_compaction_config_refresh_interval_sec,
415                hummock_version_checkpoint_interval_sec: config
416                    .meta
417                    .hummock_version_checkpoint_interval_sec,
418                enable_hummock_data_archive: config.meta.enable_hummock_data_archive,
419                checkpoint_compression_algorithm: config.meta.checkpoint_compression_algorithm,
420                checkpoint_read_chunk_size: config.meta.checkpoint_read_chunk_size,
421                checkpoint_read_max_in_flight_chunks: config
422                    .meta
423                    .checkpoint_read_max_in_flight_chunks,
424                hummock_time_travel_snapshot_interval: config
425                    .meta
426                    .hummock_time_travel_snapshot_interval,
427                hummock_time_travel_sst_info_fetch_batch_size: config
428                    .meta
429                    .developer
430                    .hummock_time_travel_sst_info_fetch_batch_size,
431                hummock_time_travel_sst_info_insert_batch_size: config
432                    .meta
433                    .developer
434                    .hummock_time_travel_sst_info_insert_batch_size,
435                hummock_time_travel_epoch_version_insert_batch_size: config
436                    .meta
437                    .developer
438                    .hummock_time_travel_epoch_version_insert_batch_size,
439                hummock_time_travel_delta_fetch_batch_size: config
440                    .meta
441                    .developer
442                    .hummock_time_travel_delta_fetch_batch_size,
443                hummock_gc_history_insert_batch_size: config
444                    .meta
445                    .developer
446                    .hummock_gc_history_insert_batch_size,
447                hummock_time_travel_filter_out_objects_batch_size: config
448                    .meta
449                    .developer
450                    .hummock_time_travel_filter_out_objects_batch_size,
451                hummock_time_travel_filter_out_objects_v1: config
452                    .meta
453                    .developer
454                    .hummock_time_travel_filter_out_objects_v1,
455                hummock_time_travel_filter_out_objects_list_version_batch_size: config
456                    .meta
457                    .developer
458                    .hummock_time_travel_filter_out_objects_list_version_batch_size,
459                hummock_time_travel_filter_out_objects_list_delta_batch_size: config
460                    .meta
461                    .developer
462                    .hummock_time_travel_filter_out_objects_list_delta_batch_size,
463                min_delta_log_num_for_hummock_version_checkpoint: config
464                    .meta
465                    .min_delta_log_num_for_hummock_version_checkpoint,
466                min_sst_retention_time_sec: config.meta.min_sst_retention_time_sec,
467                full_gc_interval_sec: config.meta.full_gc_interval_sec,
468                full_gc_object_limit: config.meta.full_gc_object_limit,
469                gc_history_retention_time_sec: config.meta.gc_history_retention_time_sec,
470                max_inflight_time_travel_query: config.meta.max_inflight_time_travel_query,
471                enable_committed_sst_sanity_check: config.meta.enable_committed_sst_sanity_check,
472                periodic_compaction_interval_sec: config.meta.periodic_compaction_interval_sec,
473                node_num_monitor_interval_sec: config.meta.node_num_monitor_interval_sec,
474                prometheus_endpoint: opts.prometheus_endpoint,
475                prometheus_selector: opts.prometheus_selector,
476                vpc_id: opts.vpc_id,
477                security_group_id: opts.security_group_id,
478                privatelink_endpoint_default_tags,
479                periodic_space_reclaim_compaction_interval_sec: config
480                    .meta
481                    .periodic_space_reclaim_compaction_interval_sec,
482                telemetry_enabled: config.server.telemetry_enabled,
483                periodic_ttl_reclaim_compaction_interval_sec: config
484                    .meta
485                    .periodic_ttl_reclaim_compaction_interval_sec,
486                periodic_tombstone_reclaim_compaction_interval_sec: config
487                    .meta
488                    .periodic_tombstone_reclaim_compaction_interval_sec,
489                periodic_scheduling_compaction_group_split_interval_sec: config
490                    .meta
491                    .periodic_scheduling_compaction_group_split_interval_sec,
492                enable_compaction_group_normalize: config.meta.enable_compaction_group_normalize,
493                max_normalize_splits_per_round: config.meta.max_normalize_splits_per_round,
494                periodic_scheduling_compaction_group_merge_interval_sec: config
495                    .meta
496                    .periodic_scheduling_compaction_group_merge_interval_sec,
497                compaction_group_merge_dimension_threshold: config
498                    .meta
499                    .compaction_group_merge_dimension_threshold,
500                table_high_write_throughput_threshold: config
501                    .meta
502                    .table_high_write_throughput_threshold,
503                table_low_write_throughput_threshold: config
504                    .meta
505                    .table_low_write_throughput_threshold,
506                partition_vnode_count: config.meta.partition_vnode_count,
507                compact_task_table_size_partition_threshold_low: config
508                    .meta
509                    .compact_task_table_size_partition_threshold_low,
510                compact_task_table_size_partition_threshold_high: config
511                    .meta
512                    .compact_task_table_size_partition_threshold_high,
513                do_not_config_object_storage_lifecycle: config
514                    .meta
515                    .do_not_config_object_storage_lifecycle,
516                compaction_task_max_heartbeat_interval_secs: config
517                    .meta
518                    .compaction_task_max_heartbeat_interval_secs,
519                compaction_task_max_progress_interval_secs,
520                compaction_task_id_refill_capacity: config.meta.compaction_task_id_refill_capacity,
521                compaction_config: Some(config.meta.compaction_config),
522                hybrid_partition_node_count: config.meta.hybrid_partition_vnode_count,
523                event_log_enabled: config.meta.event_log_enabled,
524                event_log_channel_max_size: config.meta.event_log_channel_max_size,
525                advertise_addr: opts.advertise_addr,
526                cached_traces_num: config.meta.developer.cached_traces_num,
527                cached_traces_memory_limit_bytes: config
528                    .meta
529                    .developer
530                    .cached_traces_memory_limit_bytes,
531                enable_trivial_move: config.meta.developer.enable_trivial_move,
532                enable_check_task_level_overlap: config
533                    .meta
534                    .developer
535                    .enable_check_task_level_overlap,
536                enable_dropped_column_reclaim: config.meta.enable_dropped_column_reclaim,
537                split_group_size_ratio: config.meta.split_group_size_ratio,
538                refresh_scheduler_interval_sec: config
539                    .streaming
540                    .developer
541                    .refresh_scheduler_interval_sec,
542                table_stat_high_write_throughput_ratio_for_split: config
543                    .meta
544                    .table_stat_high_write_throughput_ratio_for_split,
545                table_stat_low_write_throughput_ratio_for_merge: config
546                    .meta
547                    .table_stat_low_write_throughput_ratio_for_merge,
548                table_stat_throuput_window_seconds_for_split: config
549                    .meta
550                    .table_stat_throuput_window_seconds_for_split,
551                table_stat_throuput_window_seconds_for_merge: config
552                    .meta
553                    .table_stat_throuput_window_seconds_for_merge,
554                object_store_config: config.storage.object_store,
555                max_trivial_move_task_count_per_loop: config
556                    .meta
557                    .developer
558                    .max_trivial_move_task_count_per_loop,
559                max_get_task_probe_times: config.meta.developer.max_get_task_probe_times,
560                secret_store_private_key,
561                temp_secret_file_dir: opts.temp_secret_file_dir,
562                actor_cnt_per_worker_parallelism_hard_limit: config
563                    .meta
564                    .developer
565                    .actor_cnt_per_worker_parallelism_hard_limit,
566                actor_cnt_per_worker_parallelism_soft_limit: config
567                    .meta
568                    .developer
569                    .actor_cnt_per_worker_parallelism_soft_limit,
570                table_change_log_insert_batch_size: config
571                    .meta
572                    .developer
573                    .table_change_log_insert_batch_size,
574                table_change_log_delete_batch_size: config
575                    .meta
576                    .developer
577                    .table_change_log_delete_batch_size,
578                table_change_log_truncate_interval_sec: config
579                    .meta
580                    .developer
581                    .table_change_log_truncate_interval_sec,
582                license_key_path: opts.license_key_path,
583                compute_client_config: config.meta.developer.compute_client_config.clone(),
584                stream_client_config: config.meta.developer.stream_client_config.clone(),
585                frontend_client_config: config.meta.developer.frontend_client_config.clone(),
586                redact_sql_option_keywords: Arc::new(
587                    config
588                        .batch
589                        .redact_sql_option_keywords
590                        .into_iter()
591                        .collect(),
592                ),
593                cdc_table_split_init_sleep_interval_splits: config
594                    .meta
595                    .cdc_table_split_init_sleep_interval_splits,
596                cdc_table_split_init_sleep_duration_millis: config
597                    .meta
598                    .cdc_table_split_init_sleep_duration_millis,
599                cdc_table_split_init_insert_batch_size: config
600                    .meta
601                    .cdc_table_split_init_insert_batch_size,
602
603                enable_legacy_table_migration: config.meta.enable_legacy_table_migration,
604                pause_on_next_bootstrap_offline: config.meta.pause_on_next_bootstrap_offline,
605                serverless_backfill_controller_addr: opts.serverless_backfill_controller_addr,
606            },
607            config.system.into_init_system_params(),
608            config.session_init,
609            shutdown,
610        ))
611        .await
612        .unwrap();
613    })
614}
615
616fn validate_config(config: &RwConfig) {
617    if config.meta.meta_leader_lease_secs <= 2 {
618        let error_msg = "`meta_leader_lease_secs` must be greater than 2";
619        tracing::error!(error_msg);
620        panic!("{}", error_msg);
621    }
622
623    if config.meta.parallelism_control_batch_size == 0 {
624        let error_msg = "`parallelism_control_batch_size` must be greater than 0";
625        tracing::error!(error_msg);
626        panic!("{}", error_msg);
627    }
628
629    if config.meta.checkpoint_read_chunk_size == 0 {
630        let error_msg = "`checkpoint_read_chunk_size` must be greater than 0";
631        tracing::error!(error_msg);
632        panic!("{}", error_msg);
633    }
634
635    if config.meta.checkpoint_read_max_in_flight_chunks == 0 {
636        let error_msg = "`checkpoint_read_max_in_flight_chunks` must be greater than 0";
637        tracing::error!(error_msg);
638        panic!("{}", error_msg);
639    }
640
641    if config.meta.compaction_task_id_refill_capacity == 0 {
642        let error_msg = "`compaction_task_id_refill_capacity` must be greater than 0";
643        tracing::error!(error_msg);
644        panic!("{}", error_msg);
645    }
646
647    if config.meta.iceberg_compaction_report_timeout_sec == 0 {
648        let error_msg = "`iceberg_compaction_report_timeout_sec` must be greater than 0";
649        tracing::error!(error_msg);
650        panic!("{}", error_msg);
651    }
652
653    if config.meta.iceberg_compaction_config_refresh_interval_sec == 0 {
654        let error_msg = "`iceberg_compaction_config_refresh_interval_sec` must be greater than 0";
655        tracing::error!(error_msg);
656        panic!("{}", error_msg);
657    }
658}
risingwave_meta_node/lib.rs

risingwave_meta_node/
lib.rs
