Update error msg, enable backend config in create_sink & lib.rs

risingwavelabs · Feb 20, 2024 · 8839317 · 8839317
1 parent 1536204
commit 8839317
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 30 deletions.
diff --git a/src/frontend/src/handler/create_sink.rs b/src/frontend/src/handler/create_sink.rs
@@ -504,7 +504,7 @@ fn check_cycle_for_sink(
             if let Ok(table) = reader.get_table_by_id(table_id) {
                 visit_table(session, reader, sink_index, table.as_ref(), visited_tables)?
             } else {
-                bail!("table not found: {:?}", table_id);
+                bail!("streaming job not found: {:?}", table_id);
             }
         }
 
@@ -537,7 +537,7 @@ fn check_cycle_for_sink(
             if let Ok(table) = reader.get_table_by_id(table_id) {
                 visit_table(session, reader, sink_index, table.as_ref(), visited_tables)?
             } else {
-                bail!("table not found: {:?}", table_id);
+                bail!("streaming job not found: {:?}", table_id);
             }
         }
 

diff --git a/src/meta/node/src/lib.rs b/src/meta/node/src/lib.rs
@@ -201,30 +201,25 @@ pub fn start(opts: MetaNodeOpts) -> Pin<Box<dyn Future<Output = ()> + Send>> {
         let listen_addr = opts.listen_addr.parse().unwrap();
         let dashboard_addr = opts.dashboard_host.map(|x| x.parse().unwrap());
         let prometheus_addr = opts.prometheus_listener_addr.map(|x| x.parse().unwrap());
-        // let backend = match config.meta.backend {
-        //     MetaBackend::Etcd => MetaStoreBackend::Etcd {
-        //         endpoints: opts
-        //             .etcd_endpoints
-        //             .split(',')
-        //             .map(|x| x.to_string())
-        //             .collect(),
-        //         credentials: match opts.etcd_auth {
-        //             true => Some((
-        //                 opts.etcd_username,
-        //                 opts.etcd_password.expose_secret().to_string(),
-        //             )),
-        //             false => None,
-        //         },
-        //     },
-        //     MetaBackend::Mem => MetaStoreBackend::Mem,
-        //     MetaBackend::Sql => MetaStoreBackend::Sql {
-        //         endpoint: opts.sql_endpoint.expect("sql endpoint is required"),
-        //     },
-        // };
-        //
-
-        let backend = MetaStoreBackend::Sql {
-            endpoint: "postgres://shanicky@localhost:5432/postgres".to_string(),
+        let backend = match config.meta.backend {
+            MetaBackend::Etcd => MetaStoreBackend::Etcd {
+                endpoints: opts
+                    .etcd_endpoints
+                    .split(',')
+                    .map(|x| x.to_string())
+                    .collect(),
+                credentials: match opts.etcd_auth {
+                    true => Some((
+                        opts.etcd_username,
+                        opts.etcd_password.expose_secret().to_string(),
+                    )),
+                    false => None,
+                },
+            },
+            MetaBackend::Mem => MetaStoreBackend::Mem,
+            MetaBackend::Sql => MetaStoreBackend::Sql {
+                endpoint: opts.sql_endpoint.expect("sql endpoint is required"),
+            },
         };
 
         validate_config(&config);
@@ -285,7 +280,7 @@ pub fn start(opts: MetaNodeOpts) -> Pin<Box<dyn Future<Output = ()> + Send>> {
             max_heartbeat_interval,
             config.meta.meta_leader_lease_secs,
             MetaOpts {
-                enable_recovery: true,
+                enable_recovery: !config.meta.disable_recovery,
                 disable_automatic_parallelism_control: config
                     .meta
                     .disable_automatic_parallelism_control,

diff --git a/src/meta/src/controller/streaming_job.rs b/src/meta/src/controller/streaming_job.rs
@@ -128,7 +128,6 @@ impl CatalogController {
                     .ok_or_else(|| MetaError::catalog_id_not_found("table", object_id))?;
 
                 for sink_id in table.incoming_sinks.inner_ref() {
-                    println!("insert table {} incoming {} ", object_id, sink_id);
                     queue.push_front((*sink_id, ObjectType::Sink));
                 }
             }
@@ -146,8 +145,6 @@ impl CatalogController {
                 .all(txn)
                 .await?;
 
-            println!("deps {:?}", dependent_relations);
-
             queue.extend(dependent_relations.into_iter());
         }