risingwave_meta/manager/
env.rs

1// Copyright 2022 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
15use std::ops::Deref;
16use std::path::PathBuf;
17use std::sync::Arc;
18use std::sync::atomic::AtomicU32;
19
20use anyhow::Context;
21use risingwave_common::config::{
22    CompactionConfig, DefaultParallelism, ObjectStoreConfig, RpcClientConfig,
23};
24use risingwave_common::session_config::SessionConfig;
25use risingwave_common::system_param::reader::SystemParamsReader;
26use risingwave_common::{bail, system_param};
27use risingwave_meta_model::prelude::Cluster;
28use risingwave_pb::meta::SystemParams;
29use risingwave_rpc_client::{
30    FrontendClientPool, FrontendClientPoolRef, StreamClientPool, StreamClientPoolRef,
31};
32use risingwave_sqlparser::ast::RedactSqlOptionKeywordsRef;
33use sea_orm::EntityTrait;
34
35use crate::MetaResult;
36use crate::barrier::SharedActorInfos;
37use crate::controller::SqlMetaStore;
38use crate::controller::id::{
39    IdGeneratorManager as SqlIdGeneratorManager, IdGeneratorManagerRef as SqlIdGeneratorManagerRef,
40};
41use crate::controller::session_params::{SessionParamsController, SessionParamsControllerRef};
42use crate::controller::system_param::{SystemParamsController, SystemParamsControllerRef};
43use crate::hummock::sequence::SequenceGenerator;
44use crate::manager::event_log::{EventLogManagerRef, start_event_log_manager};
45use crate::manager::{IdleManager, IdleManagerRef, NotificationManager, NotificationManagerRef};
46use crate::model::ClusterId;
47
48/// [`MetaSrvEnv`] is the global environment in Meta service. The instance will be shared by all
49/// kind of managers inside Meta.
50#[derive(Clone)]
51pub struct MetaSrvEnv {
52    /// id generator manager.
53    id_gen_manager_impl: SqlIdGeneratorManagerRef,
54
55    /// system param manager.
56    system_param_manager_impl: SystemParamsControllerRef,
57
58    /// session param manager.
59    session_param_manager_impl: SessionParamsControllerRef,
60
61    /// meta store.
62    meta_store_impl: SqlMetaStore,
63
64    /// notification manager.
65    notification_manager: NotificationManagerRef,
66
67    pub shared_actor_info: SharedActorInfos,
68
69    /// stream client pool memorization.
70    stream_client_pool: StreamClientPoolRef,
71
72    /// rpc client pool for frontend nodes.
73    frontend_client_pool: FrontendClientPoolRef,
74
75    /// idle status manager.
76    idle_manager: IdleManagerRef,
77
78    event_log_manager: EventLogManagerRef,
79
80    /// Unique identifier of the cluster.
81    cluster_id: ClusterId,
82
83    pub hummock_seq: Arc<SequenceGenerator>,
84
85    /// The await-tree registry of the current meta node.
86    await_tree_reg: await_tree::Registry,
87
88    /// options read by all services
89    pub opts: Arc<MetaOpts>,
90
91    actor_id_generator: Arc<AtomicU32>,
92}
93
94/// Options shared by all meta service instances
95#[derive(Clone, serde::Serialize)]
96pub struct MetaOpts {
97    /// Whether to enable the recovery of the cluster. If disabled, the meta service will exit on
98    /// abnormal cases.
99    pub enable_recovery: bool,
100    /// Whether to disable the auto-scaling feature.
101    pub disable_automatic_parallelism_control: bool,
102    /// The number of streaming jobs per scaling operation.
103    pub parallelism_control_batch_size: usize,
104    /// The period of parallelism control trigger.
105    pub parallelism_control_trigger_period_sec: u64,
106    /// The first delay of parallelism control.
107    pub parallelism_control_trigger_first_delay_sec: u64,
108    /// The maximum number of barriers in-flight in the compute nodes.
109    pub in_flight_barrier_nums: usize,
110    /// After specified seconds of idle (no mview or flush), the process will be exited.
111    /// 0 for infinite, process will never be exited due to long idle time.
112    pub max_idle_ms: u64,
113    /// Whether run in compaction detection test mode
114    pub compaction_deterministic_test: bool,
115    /// Default parallelism of units for all streaming jobs.
116    pub default_parallelism: DefaultParallelism,
117
118    /// Interval of invoking a vacuum job, to remove stale metadata from meta store and objects
119    /// from object store.
120    pub vacuum_interval_sec: u64,
121    /// The spin interval inside a vacuum job. It avoids the vacuum job monopolizing resources of
122    /// meta node.
123    pub vacuum_spin_interval_ms: u64,
124    /// Interval of invoking iceberg garbage collection, to expire old snapshots.
125    pub iceberg_gc_interval_sec: u64,
126    pub time_travel_vacuum_interval_sec: u64,
127    pub time_travel_vacuum_max_version_count: Option<u32>,
128    /// Interval of hummock version checkpoint.
129    pub hummock_version_checkpoint_interval_sec: u64,
130    pub enable_hummock_data_archive: bool,
131    /// Compression algorithm for hummock version checkpoint: "zstd", "lz4", or "none".
132    pub checkpoint_compression_algorithm: risingwave_common::config::CheckpointCompression,
133    /// Chunk size in bytes for reading large checkpoints.
134    pub checkpoint_read_chunk_size: usize,
135    /// Maximum number of concurrent chunk reads for large checkpoints.
136    pub checkpoint_read_max_in_flight_chunks: usize,
137    pub hummock_time_travel_snapshot_interval: u64,
138    pub hummock_time_travel_sst_info_fetch_batch_size: usize,
139    pub hummock_time_travel_sst_info_insert_batch_size: usize,
140    pub hummock_time_travel_epoch_version_insert_batch_size: usize,
141    pub hummock_time_travel_delta_fetch_batch_size: usize,
142    pub hummock_gc_history_insert_batch_size: usize,
143    pub hummock_time_travel_filter_out_objects_batch_size: usize,
144    pub hummock_time_travel_filter_out_objects_v1: bool,
145    pub hummock_time_travel_filter_out_objects_list_version_batch_size: usize,
146    pub hummock_time_travel_filter_out_objects_list_delta_batch_size: usize,
147    /// The minimum delta log number a new checkpoint should compact, otherwise the checkpoint
148    /// attempt is rejected. Greater value reduces object store IO, meanwhile it results in
149    /// more loss of in memory `HummockVersionCheckpoint::stale_objects` state when meta node is
150    /// restarted.
151    pub min_delta_log_num_for_hummock_version_checkpoint: u64,
152    /// Objects within `min_sst_retention_time_sec` won't be deleted by hummock full GC, even they
153    /// are dangling.
154    pub min_sst_retention_time_sec: u64,
155    /// Interval of automatic hummock full GC.
156    pub full_gc_interval_sec: u64,
157    /// Max number of object per full GC job can fetch.
158    pub full_gc_object_limit: u64,
159    /// Duration in seconds to retain garbage collection history data.
160    pub gc_history_retention_time_sec: u64,
161    /// Max number of inflight time travel query.
162    pub max_inflight_time_travel_query: u64,
163    /// Enable sanity check when SSTs are committed
164    pub enable_committed_sst_sanity_check: bool,
165    /// Schedule compaction for all compaction groups with this interval.
166    pub periodic_compaction_interval_sec: u64,
167    /// Interval of reporting the number of nodes in the cluster.
168    pub node_num_monitor_interval_sec: u64,
169    /// Whether to protect the drop table operation with incoming sink.
170    pub protect_drop_table_with_incoming_sink: bool,
171    /// The Prometheus endpoint for Meta Dashboard Service.
172    /// The Dashboard service uses this in the following ways:
173    /// 1. Query Prometheus for relevant metrics to find Stream Graph Bottleneck, and display it.
174    /// 2. Provide cluster diagnostics, at `/api/monitor/diagnose` to troubleshoot cluster.
175    ///    These are just examples which show how the Meta Dashboard Service queries Prometheus.
176    pub prometheus_endpoint: Option<String>,
177
178    /// The additional selector used when querying Prometheus.
179    pub prometheus_selector: Option<String>,
180
181    /// The VPC id of the cluster.
182    pub vpc_id: Option<String>,
183
184    /// A usable security group id to assign to a vpc endpoint
185    pub security_group_id: Option<String>,
186
187    /// Default tag for the endpoint created when creating a privatelink connection.
188    /// Will be appended to the tags specified in the `tags` field in with clause in `create
189    /// connection`.
190    pub privatelink_endpoint_default_tags: Option<Vec<(String, String)>>,
191
192    /// Schedule `space_reclaim_compaction` for all compaction groups with this interval.
193    pub periodic_space_reclaim_compaction_interval_sec: u64,
194
195    /// telemetry enabled in config file or not
196    pub telemetry_enabled: bool,
197    /// Schedule `ttl_reclaim_compaction` for all compaction groups with this interval.
198    pub periodic_ttl_reclaim_compaction_interval_sec: u64,
199
200    /// Schedule `tombstone_reclaim_compaction` for all compaction groups with this interval.
201    pub periodic_tombstone_reclaim_compaction_interval_sec: u64,
202
203    /// Schedule the regular compaction-group split job for all compaction groups with this interval.
204    pub periodic_scheduling_compaction_group_split_interval_sec: u64,
205    /// Whether to enable overlap normalization before the regular merge scheduler.
206    pub enable_compaction_group_normalize: bool,
207    /// Maximum normalize splits in one scheduler round. Must be greater than 0.
208    pub max_normalize_splits_per_round: u64,
209
210    /// Whether config object storage bucket lifecycle to purge stale data.
211    pub do_not_config_object_storage_lifecycle: bool,
212
213    pub partition_vnode_count: u32,
214
215    /// threshold of high write throughput of state-table, unit: B/sec
216    pub table_high_write_throughput_threshold: u64,
217    /// threshold of low write throughput of state-table, unit: B/sec
218    pub table_low_write_throughput_threshold: u64,
219
220    pub compaction_task_max_heartbeat_interval_secs: u64,
221    pub compaction_task_max_progress_interval_secs: u64,
222    pub compaction_task_id_refill_capacity: u32,
223    pub compaction_config: Option<CompactionConfig>,
224
225    /// hybrid compaction group config
226    ///
227    /// `hybrid_partition_vnode_count` determines the granularity of vnodes in the hybrid compaction group for SST alignment.
228    /// When `hybrid_partition_vnode_count` > 0, in hybrid compaction group
229    /// - Tables with high write throughput will be split at vnode granularity
230    /// - Tables with high size tables will be split by table granularity
231    ///   When `hybrid_partition_vnode_count` = 0,no longer be special alignment operations for the hybrid compaction group
232    pub hybrid_partition_node_count: u32,
233
234    pub event_log_enabled: bool,
235    pub event_log_channel_max_size: u32,
236    pub advertise_addr: String,
237    /// The number of traces to be cached in-memory by the tracing collector
238    /// embedded in the meta node.
239    pub cached_traces_num: u32,
240    /// The maximum memory usage in bytes for the tracing collector embedded
241    /// in the meta node.
242    pub cached_traces_memory_limit_bytes: usize,
243
244    /// l0 picker whether to select trivial move task
245    pub enable_trivial_move: bool,
246
247    /// l0 multi level picker whether to check the overlap accuracy between sub levels
248    pub enable_check_task_level_overlap: bool,
249    pub enable_dropped_column_reclaim: bool,
250
251    /// Whether to split the compaction group when the size of the group exceeds the threshold.
252    pub split_group_size_ratio: f64,
253
254    /// The interval in seconds for the refresh scheduler to check and trigger scheduled refreshes.
255    pub refresh_scheduler_interval_sec: u64,
256
257    /// To split the compaction group when the high throughput statistics of the group exceeds the threshold.
258    pub table_stat_high_write_throughput_ratio_for_split: f64,
259
260    /// To merge the compaction group when the low throughput statistics of the group exceeds the threshold.
261    pub table_stat_low_write_throughput_ratio_for_merge: f64,
262
263    /// The window seconds of table throughput statistic history for split compaction group.
264    pub table_stat_throuput_window_seconds_for_split: usize,
265
266    /// The window seconds of table throughput statistic history for merge compaction group.
267    pub table_stat_throuput_window_seconds_for_merge: usize,
268
269    /// The configuration of the object store
270    pub object_store_config: ObjectStoreConfig,
271
272    /// The maximum number of trivial move tasks to be picked in a single loop
273    pub max_trivial_move_task_count_per_loop: usize,
274
275    /// The maximum number of times to probe for `PullTaskEvent`
276    pub max_get_task_probe_times: usize,
277
278    pub compact_task_table_size_partition_threshold_low: u64,
279    pub compact_task_table_size_partition_threshold_high: u64,
280
281    pub periodic_scheduling_compaction_group_merge_interval_sec: u64,
282
283    pub compaction_group_merge_dimension_threshold: f64,
284
285    // The private key for the secret store, used when the secret is stored in the meta.
286    pub secret_store_private_key: Option<Vec<u8>>,
287    /// The path of the temp secret file directory.
288    pub temp_secret_file_dir: String,
289
290    // Cluster limits
291    pub actor_cnt_per_worker_parallelism_hard_limit: usize,
292    pub actor_cnt_per_worker_parallelism_soft_limit: usize,
293
294    pub table_change_log_insert_batch_size: u64,
295    pub table_change_log_delete_batch_size: u64,
296
297    pub license_key_path: Option<PathBuf>,
298
299    pub compute_client_config: RpcClientConfig,
300    pub stream_client_config: RpcClientConfig,
301    pub frontend_client_config: RpcClientConfig,
302    pub redact_sql_option_keywords: RedactSqlOptionKeywordsRef,
303
304    pub cdc_table_split_init_sleep_interval_splits: u64,
305    pub cdc_table_split_init_sleep_duration_millis: u64,
306    pub cdc_table_split_init_insert_batch_size: u64,
307
308    pub enable_legacy_table_migration: bool,
309    pub pause_on_next_bootstrap_offline: bool,
310}
311
312impl MetaOpts {
313    /// Default opts for testing. Some tests need `enable_recovery=true`
314    pub fn test(enable_recovery: bool) -> Self {
315        Self {
316            enable_recovery,
317            disable_automatic_parallelism_control: false,
318            parallelism_control_batch_size: 1,
319            parallelism_control_trigger_period_sec: 10,
320            parallelism_control_trigger_first_delay_sec: 30,
321            in_flight_barrier_nums: 40,
322            max_idle_ms: 0,
323            compaction_deterministic_test: false,
324            default_parallelism: DefaultParallelism::Full,
325            vacuum_interval_sec: 30,
326            time_travel_vacuum_interval_sec: 30,
327            time_travel_vacuum_max_version_count: None,
328            vacuum_spin_interval_ms: 0,
329            iceberg_gc_interval_sec: 3600,
330            hummock_version_checkpoint_interval_sec: 30,
331            enable_hummock_data_archive: false,
332            checkpoint_compression_algorithm:
333                risingwave_common::config::CheckpointCompression::Zstd,
334            checkpoint_read_chunk_size: 128 * 1024 * 1024,
335            checkpoint_read_max_in_flight_chunks: 4,
336            hummock_time_travel_snapshot_interval: 0,
337            hummock_time_travel_sst_info_fetch_batch_size: 10_000,
338            hummock_time_travel_sst_info_insert_batch_size: 10,
339            hummock_time_travel_epoch_version_insert_batch_size: 1000,
340            hummock_time_travel_delta_fetch_batch_size: 1000,
341            hummock_gc_history_insert_batch_size: 1000,
342            hummock_time_travel_filter_out_objects_batch_size: 1000,
343            hummock_time_travel_filter_out_objects_v1: false,
344            hummock_time_travel_filter_out_objects_list_version_batch_size: 10,
345            hummock_time_travel_filter_out_objects_list_delta_batch_size: 1000,
346            min_delta_log_num_for_hummock_version_checkpoint: 1,
347            min_sst_retention_time_sec: 3600 * 24 * 7,
348            full_gc_interval_sec: 3600 * 24 * 7,
349            full_gc_object_limit: 100_000,
350            gc_history_retention_time_sec: 3600 * 24 * 7,
351            max_inflight_time_travel_query: 1000,
352            enable_committed_sst_sanity_check: false,
353            periodic_compaction_interval_sec: 300,
354            node_num_monitor_interval_sec: 10,
355            protect_drop_table_with_incoming_sink: false,
356            prometheus_endpoint: None,
357            prometheus_selector: None,
358            vpc_id: None,
359            security_group_id: None,
360            privatelink_endpoint_default_tags: None,
361            periodic_space_reclaim_compaction_interval_sec: 60,
362            telemetry_enabled: false,
363            periodic_ttl_reclaim_compaction_interval_sec: 60,
364            periodic_tombstone_reclaim_compaction_interval_sec: 60,
365            periodic_scheduling_compaction_group_split_interval_sec: 60,
366            enable_compaction_group_normalize: false,
367            max_normalize_splits_per_round: 4,
368            compact_task_table_size_partition_threshold_low: 128 * 1024 * 1024,
369            compact_task_table_size_partition_threshold_high: 512 * 1024 * 1024,
370            table_high_write_throughput_threshold: 128 * 1024 * 1024,
371            table_low_write_throughput_threshold: 64 * 1024 * 1024,
372            do_not_config_object_storage_lifecycle: true,
373            partition_vnode_count: 32,
374            compaction_task_max_heartbeat_interval_secs: 0,
375            compaction_task_max_progress_interval_secs: 1,
376            compaction_task_id_refill_capacity: 64,
377            compaction_config: None,
378            hybrid_partition_node_count: 4,
379            event_log_enabled: false,
380            event_log_channel_max_size: 1,
381            advertise_addr: "".to_owned(),
382            cached_traces_num: 1,
383            cached_traces_memory_limit_bytes: usize::MAX,
384            enable_trivial_move: true,
385            enable_check_task_level_overlap: true,
386            enable_dropped_column_reclaim: false,
387            object_store_config: ObjectStoreConfig::default(),
388            max_trivial_move_task_count_per_loop: 256,
389            max_get_task_probe_times: 5,
390            secret_store_private_key: Some(
391                hex::decode("0123456789abcdef0123456789abcdef").unwrap(),
392            ),
393            temp_secret_file_dir: "./secrets".to_owned(),
394            actor_cnt_per_worker_parallelism_hard_limit: usize::MAX,
395            actor_cnt_per_worker_parallelism_soft_limit: usize::MAX,
396            split_group_size_ratio: 0.9,
397            table_stat_high_write_throughput_ratio_for_split: 0.5,
398            table_stat_low_write_throughput_ratio_for_merge: 0.7,
399            table_stat_throuput_window_seconds_for_split: 60,
400            table_stat_throuput_window_seconds_for_merge: 240,
401            periodic_scheduling_compaction_group_merge_interval_sec: 60 * 10,
402            compaction_group_merge_dimension_threshold: 1.2,
403            license_key_path: None,
404            compute_client_config: RpcClientConfig::default(),
405            stream_client_config: RpcClientConfig::default(),
406            frontend_client_config: RpcClientConfig::default(),
407            redact_sql_option_keywords: Arc::new(Default::default()),
408            cdc_table_split_init_sleep_interval_splits: 1000,
409            cdc_table_split_init_sleep_duration_millis: 10,
410            cdc_table_split_init_insert_batch_size: 1000,
411            enable_legacy_table_migration: true,
412            refresh_scheduler_interval_sec: 60,
413            pause_on_next_bootstrap_offline: false,
414            table_change_log_insert_batch_size: 1000,
415            table_change_log_delete_batch_size: 1000,
416        }
417    }
418}
419
420impl MetaSrvEnv {
421    pub async fn new(
422        opts: MetaOpts,
423        mut init_system_params: SystemParams,
424        init_session_config: SessionConfig,
425        meta_store_impl: SqlMetaStore,
426    ) -> MetaResult<Self> {
427        let idle_manager = Arc::new(IdleManager::new(opts.max_idle_ms));
428        let stream_client_pool =
429            Arc::new(StreamClientPool::new(1, opts.stream_client_config.clone())); // typically no need for plural clients
430        let frontend_client_pool = Arc::new(FrontendClientPool::new(
431            1,
432            opts.frontend_client_config.clone(),
433        ));
434        let event_log_manager = Arc::new(start_event_log_manager(
435            opts.event_log_enabled,
436            opts.event_log_channel_max_size,
437        ));
438
439        // When license key path is specified, license key from system parameters can be easily
440        // overwritten. So we simply reject this case.
441        if opts.license_key_path.is_some()
442            && init_system_params.license_key
443                != system_param::default::license_key_opt().map(Into::into)
444        {
445            bail!(
446                "argument `--license-key-path` (or env var `RW_LICENSE_KEY_PATH`) and \
447                 system parameter `license_key` (or env var `RW_LICENSE_KEY`) may not \
448                 be set at the same time"
449            );
450        }
451
452        let cluster_first_launch = meta_store_impl.up().await.context(
453            "Failed to initialize the meta store, \
454            this may happen if there's existing metadata incompatible with the current version of RisingWave, \
455            e.g., downgrading from a newer release or a nightly build to an older one. \
456            For a single-node deployment, you may want to reset all data by deleting the data directory, \
457            typically located at `~/.risingwave`.",
458        )?;
459
460        let notification_manager =
461            Arc::new(NotificationManager::new(meta_store_impl.clone()).await);
462        let cluster_id = Cluster::find()
463            .one(&meta_store_impl.conn)
464            .await?
465            .map(|c| c.cluster_id.to_string().into())
466            .unwrap();
467
468        // For new clusters:
469        // - the name of the object store needs to be prefixed according to the object id.
470        //
471        // For old clusters
472        // - the prefix is not divided for the sake of compatibility.
473        init_system_params.use_new_object_prefix_strategy = Some(cluster_first_launch);
474
475        let system_param_controller = Arc::new(
476            SystemParamsController::new(
477                meta_store_impl.clone(),
478                notification_manager.clone(),
479                init_system_params,
480            )
481            .await?,
482        );
483        let session_param_controller = Arc::new(
484            SessionParamsController::new(
485                meta_store_impl.clone(),
486                notification_manager.clone(),
487                init_session_config,
488            )
489            .await?,
490        );
491        Ok(Self {
492            id_gen_manager_impl: Arc::new(SqlIdGeneratorManager::new(&meta_store_impl.conn).await?),
493            system_param_manager_impl: system_param_controller,
494            session_param_manager_impl: session_param_controller,
495            meta_store_impl: meta_store_impl.clone(),
496            shared_actor_info: SharedActorInfos::new(notification_manager.clone()),
497            notification_manager,
498            stream_client_pool,
499            frontend_client_pool,
500            idle_manager,
501            event_log_manager,
502            cluster_id,
503            hummock_seq: Arc::new(SequenceGenerator::new(meta_store_impl.conn.clone())),
504            opts: opts.into(),
505            // Await trees on the meta node is lightweight, thus always enabled.
506            await_tree_reg: await_tree::Registry::new(Default::default()),
507            actor_id_generator: Arc::new(AtomicU32::new(0)),
508        })
509    }
510
511    pub fn meta_store(&self) -> SqlMetaStore {
512        self.meta_store_impl.clone()
513    }
514
515    pub fn meta_store_ref(&self) -> &SqlMetaStore {
516        &self.meta_store_impl
517    }
518
519    pub fn id_gen_manager(&self) -> &SqlIdGeneratorManagerRef {
520        &self.id_gen_manager_impl
521    }
522
523    pub fn notification_manager_ref(&self) -> NotificationManagerRef {
524        self.notification_manager.clone()
525    }
526
527    pub fn notification_manager(&self) -> &NotificationManager {
528        self.notification_manager.deref()
529    }
530
531    pub fn idle_manager_ref(&self) -> IdleManagerRef {
532        self.idle_manager.clone()
533    }
534
535    pub fn idle_manager(&self) -> &IdleManager {
536        self.idle_manager.deref()
537    }
538
539    pub fn actor_id_generator(&self) -> &AtomicU32 {
540        self.actor_id_generator.deref()
541    }
542
543    pub async fn system_params_reader(&self) -> SystemParamsReader {
544        self.system_param_manager_impl.get_params().await
545    }
546
547    pub fn system_params_manager_impl_ref(&self) -> SystemParamsControllerRef {
548        self.system_param_manager_impl.clone()
549    }
550
551    pub fn session_params_manager_impl_ref(&self) -> SessionParamsControllerRef {
552        self.session_param_manager_impl.clone()
553    }
554
555    pub fn stream_client_pool_ref(&self) -> StreamClientPoolRef {
556        self.stream_client_pool.clone()
557    }
558
559    pub fn stream_client_pool(&self) -> &StreamClientPool {
560        self.stream_client_pool.deref()
561    }
562
563    pub fn frontend_client_pool(&self) -> &FrontendClientPool {
564        self.frontend_client_pool.deref()
565    }
566
567    pub fn cluster_id(&self) -> &ClusterId {
568        &self.cluster_id
569    }
570
571    pub fn event_log_manager_ref(&self) -> EventLogManagerRef {
572        self.event_log_manager.clone()
573    }
574
575    pub fn await_tree_reg(&self) -> &await_tree::Registry {
576        &self.await_tree_reg
577    }
578
579    pub fn shared_actor_infos(&self) -> &SharedActorInfos {
580        &self.shared_actor_info
581    }
582}
583
584#[cfg(any(test, feature = "test"))]
585impl MetaSrvEnv {
586    // Instance for test.
587    pub async fn for_test() -> Self {
588        Self::for_test_opts(MetaOpts::test(false), |_| ()).await
589    }
590
591    pub async fn for_test_opts(
592        opts: MetaOpts,
593        on_test_system_params: impl FnOnce(&mut risingwave_pb::meta::PbSystemParams),
594    ) -> Self {
595        let mut system_params = risingwave_common::system_param::system_params_for_test();
596        on_test_system_params(&mut system_params);
597        Self::new(
598            opts,
599            system_params,
600            Default::default(),
601            SqlMetaStore::for_test().await,
602        )
603        .await
604        .unwrap()
605    }
606}
risingwave_meta/manager/env.rs

risingwave_meta/manager/
env.rs
