add config for easy of testing

Makefile.toml

@@ -86,9 +86,9 @@ if ${is_not_ci}
   no_rust_log = not ${rust_log}
 
   if ${no_rust_log}
-    set_env RUST_LOG "pgwire_query_log=info"
+    set_env RUST_LOG "pgwire_query_log=info,hyper::client::connect::http=info"
   else
-    set_env RUST_LOG "pgwire_query_log=info,${rust_log}"
+    set_env RUST_LOG "pgwire_query_log=info,hyper::client::connect::http=info,${rust_log}"
   end
 end
 ''',

src/common/src/lib.rs

@@ -44,6 +44,8 @@
 #![feature(bound_map)]
 #![feature(array_methods)]
 #![feature(btree_cursors)]
+#![feature(register_tool)]
+#![register_tool(rw)]
 
 #[cfg_attr(not(test), allow(unused_extern_crates))]
 extern crate self as risingwave_common;

src/common/src/session_config/mod.rs

@@ -243,6 +243,10 @@ pub struct ConfigMap {
     #[parameter(default = false)]
     background_ddl: bool,
 
+    /// Run DDL statements in background
+    #[parameter(default = false)]
+    enable_reusable_source: bool,
+
     /// Shows the server-side character set encoding. At present, this parameter can be shown but not set, because the encoding is determined at database creation time.
     #[parameter(default = SERVER_ENCODING)]
     server_encoding: String,

src/common/src/system_param/mod.rs

@@ -317,7 +317,8 @@ macro_rules! impl_set_system_param {
                 $(
                     key_of!($field) => {
                         let v = if let Some(v) = value {
-                            v.as_ref().parse().map_err(|_| format!("cannot parse parameter value"))?
+                            #[allow(rw::format_error)]
+                            v.as_ref().parse().map_err(|e| format!("cannot parse parameter value: {e}"))?
                         } else {
                             $default.ok_or_else(|| format!("{} does not have a default value", key))?
                         };

src/connector/src/source/kafka/private_link.rs

@@ -71,9 +71,9 @@ impl BrokerAddrRewriter {
         role: PrivateLinkContextRole,
         broker_rewrite_map: Option<HashMap<String, String>>,
     ) -> ConnectorResult<Self> {
-        tracing::info!("[{}] rewrite map {:?}", role, broker_rewrite_map);
         let rewrite_map: ConnectorResult<BTreeMap<BrokerAddr, BrokerAddr>> = broker_rewrite_map
             .map_or(Ok(BTreeMap::new()), |addr_map| {
+                tracing::info!("[{}] rewrite map {:?}", role, addr_map);
                 addr_map
                     .into_iter()
                     .map(|(old_addr, new_addr)| {

src/frontend/src/handler/create_source.rs

@@ -1417,7 +1417,7 @@ pub async fn handle_create_source(
 
     let catalog_writer = session.catalog_writer()?;
 
-    if has_streaming_job {
+    if has_streaming_job && session.config().enable_reusable_source() {
         let graph = {
             let context = OptimizerContext::from_handler_args(handler_args);
             let source_node = LogicalSource::with_catalog(

src/frontend/src/planner/relation.rs

@@ -87,7 +87,7 @@ impl Planner {
     }
 
     pub(super) fn plan_source(&mut self, source: BoundSource) -> Result<PlanRef> {
-        if source.can_backfill() {
+        if source.can_backfill() && self.ctx().session_ctx().config().enable_reusable_source() {
             Ok(LogicalSourceBackfill::new(Rc::new(source.catalog), self.ctx())?.into())
         } else {
             Ok(LogicalSource::with_catalog(