feldera · mihaibudiu · Jul 3, 2026 · Jul 2, 2026
diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
@@ -4764,20 +4764,6 @@ impl ControllerInit {
             )
         }
 
-        // Transfer HTTP input endpoints that are not affected by the program diff from the checkpoint to the new configuration.
-        checkpoint_config
-            .inputs
-            .iter()
-            .filter(|(_connector_name, connector_config)| {
-                connector_config.connector_config.transport.is_http_input()
-                    && !pipeline_diff.is_affected_relation(&connector_config.stream)
-            })
-            .for_each(|(connector_name, connector_config)| {
-                config
-                    .inputs
-                    .insert(connector_name.clone(), connector_config.clone());
-            });
-
         // Merge `config` (the configuration provided by the pipeline manager)
         // with `checkpoint_config` (the configuration read from the
         // checkpoint).
@@ -4830,26 +4816,11 @@ impl ControllerInit {
                 pipeline_template_configmap: config.global.pipeline_template_configmap.clone(),
             },
 
-            // If pipeline is unmodified, we may need to replay journaled inputs.
-            // We therefore use connector configuration from the checkpoint, including
-            // transient HTTP and adhoc connectors, so that we can use them to
-            // replay journaled inputs.
-            inputs: if !modified {
-                checkpoint_config
-                    .inputs
-                    .into_iter()
-                    .filter(|(_, config)| {
-                        // The clock input connector will be automatically recreated and initialized
-                        // with the clock resolution from the pipeline config.
-                        !matches!(
-                            config.connector_config.transport,
-                            TransportConfig::ClockInput(_)
-                        )
-                    })
-                    .collect()
-            } else {
-                config.inputs
-            },
+            inputs: Self::inputs_for_checkpoint_replay(
+                config.inputs,
+                checkpoint_config.inputs,
+                &pipeline_diff,
+            ),
             outputs: if !modified {
                 checkpoint_config.outputs
             } else {
@@ -4880,6 +4851,69 @@ impl ControllerInit {
         })
     }
 
+    fn inputs_for_checkpoint_replay(
+        mut config_inputs: BTreeMap<Cow<'static, str>, InputEndpointConfig>,
+        checkpoint_inputs: BTreeMap<Cow<'static, str>, InputEndpointConfig>,
+        pipeline_diff: &feldera_types::pipeline_diff::PipelineDiff,
+    ) -> BTreeMap<Cow<'static, str>, InputEndpointConfig> {
+        // Preserve the pipeline manager's input configs before HTTP input
+        // endpoints are transferred from the checkpoint below.  The final
+        // replay config still needs replay-safe limits from the new config.
+        let new_inputs = config_inputs.clone();
+
+        // Transfer HTTP input endpoints that are not affected by the program
+        // diff from the checkpoint to the new configuration.
+        checkpoint_inputs
+            .iter()
+            .filter(|(_connector_name, connector_config)| {
+                connector_config.connector_config.transport.is_http_input()
+                    && !pipeline_diff.is_affected_relation(&connector_config.stream)
+            })
+            .for_each(|(connector_name, connector_config)| {
+                config_inputs.insert(connector_name.clone(), connector_config.clone());
+            });
+
+        if pipeline_diff.is_empty() {
+            // If the pipeline is unmodified, we may need to replay journaled
+            // inputs. We therefore use connector configuration from the
+            // checkpoint, including transient HTTP and adhoc connectors, so
+            // that we can use them to replay journaled inputs. Input
+            // flow-control settings are safe to adopt from the new config
+            // without invalidating checkpointed state.
+            Self::checkpoint_inputs_with_replay_safe_config(checkpoint_inputs, &new_inputs)
+        } else {
+            // Otherwise, keep the new pipeline-manager config, including
+            // transferred HTTP and adhoc inputs needed to replay journaled
+            // inputs after a program-diff bootstrap.
+            config_inputs
+        }
+    }
+
+    fn checkpoint_inputs_with_replay_safe_config(
+        mut checkpoint_inputs: BTreeMap<Cow<'static, str>, InputEndpointConfig>,
+        new_inputs: &BTreeMap<Cow<'static, str>, InputEndpointConfig>,
+    ) -> BTreeMap<Cow<'static, str>, InputEndpointConfig> {
+        for (connector_name, checkpoint_input) in checkpoint_inputs.iter_mut() {
+            if let Some(new_input) = new_inputs.get(connector_name) {
+                checkpoint_input
+                    .connector_config
+                    .apply_input_checkpoint_replay_config_from(&new_input.connector_config);
+            }
+        }
+
+        checkpoint_inputs
+            .into_iter()
+            .filter(|(_, config)| {
+                // The clock input connector will be automatically recreated and initialized
+                // with the clock resolution from the pipeline config.
+                !matches!(
+                    config.connector_config.transport,
+                    TransportConfig::ClockInput(_)
+                )
+            })
+            .collect()
+    }
+
     pub fn set_incarnation_uuid(&mut self, incarnation_uuid: Uuid) {
         self.incarnation_uuid = incarnation_uuid
     }
@@ -8378,7 +8412,10 @@ mod controller_init_tests {
     use super::ControllerInit;
     use crate::ControllerError;
     use feldera_adapterlib::errors::controller::ConfigError;
-    use feldera_types::config::{PipelineConfig, RuntimeConfig};
+    use feldera_types::config::{InputEndpointConfig, PipelineConfig, RuntimeConfig};
+    use feldera_types::pipeline_diff::PipelineDiff;
+    use serde_json::json;
+    use std::{borrow::Cow, collections::BTreeMap};
 
     fn pipeline_config(global: RuntimeConfig) -> PipelineConfig {
         PipelineConfig {
@@ -8442,7 +8479,85 @@ mod controller_init_tests {
                 if matches!(
                     *config_error,
                     ConfigError::DatafusionMemoryExceedsBudget { .. },
-                ),
+            ),
         ));
     }
+
+    #[test]
+    fn checkpoint_inputs_adopt_replay_safe_config() {
+        let mut checkpoint_inputs = BTreeMap::new();
+        checkpoint_inputs.insert(
+            Cow::Borrowed("test_input.connector"),
+            input_config(json!({"name": "empty_input"}), 100),
+        );
+
+        let mut new_inputs = BTreeMap::new();
+        new_inputs.insert(
+            Cow::Borrowed("test_input.connector"),
+            input_config(json!({"name": "empty_input"}), 200),
+        );
+
+        let merged = ControllerInit::checkpoint_inputs_with_replay_safe_config(
+            checkpoint_inputs,
+            &new_inputs,
+        );
+        let connector_config = &merged["test_input.connector"].connector_config;
+
+        assert_eq!(connector_config.max_queued_records, 200);
+        assert_eq!(connector_config.max_queued_bytes, Some(400));
+        assert_eq!(connector_config.max_batch_size, Some(201));
+        assert_eq!(connector_config.max_worker_batch_size, Some(202));
+    }
+
+    #[test]
+    fn checkpoint_http_input_replay_uses_pre_transfer_config() {
+        let mut checkpoint_inputs = BTreeMap::new();
+        checkpoint_inputs.insert(
+            Cow::Borrowed("test_input.connector"),
+            input_config(
+                json!({
+                    "name": "http_input",
+                    "config": {"name": "test_input.connector"}
+                }),
+                100,
+            ),
+        );
+
+        let mut config_inputs = BTreeMap::new();
+        config_inputs.insert(
+            Cow::Borrowed("test_input.connector"),
+            input_config(
+                json!({
+                    "name": "http_input",
+                    "config": {"name": "test_input.connector"}
+                }),
+                200,
+            ),
+        );
+
+        let merged = ControllerInit::inputs_for_checkpoint_replay(
+            config_inputs,
+            checkpoint_inputs,
+            &PipelineDiff::new_with_program_diff(Default::default()),
+        );
+        let connector_config = &merged["test_input.connector"].connector_config;
+
+        assert!(connector_config.transport.is_http_input());
+        assert_eq!(connector_config.max_queued_records, 200);
+        assert_eq!(connector_config.max_queued_bytes, Some(400));
+        assert_eq!(connector_config.max_batch_size, Some(201));
+        assert_eq!(connector_config.max_worker_batch_size, Some(202));
+    }
+
+    fn input_config(transport: serde_json::Value, max_queued_records: u64) -> InputEndpointConfig {
+        serde_json::from_value(json!({
+            "stream": "test_input",
+            "transport": transport,
+            "max_queued_records": max_queued_records,
+            "max_queued_bytes": max_queued_records * 2,
+            "max_batch_size": max_queued_records + 1,
+            "max_worker_batch_size": max_queued_records + 2,
+        }))
+        .unwrap()
+    }
 }
diff --git a/crates/adapters/src/controller/pipeline_diff.rs b/crates/adapters/src/controller/pipeline_diff.rs
@@ -147,7 +147,7 @@ pub fn compute_pipeline_diff(
                     .get(*k)
                     .unwrap()
                     .connector_config
-                    .equal_modulo_paused(&v.connector_config)
+                    .equal_for_input_checkpoint_replay(&v.connector_config)
         })
         .map(|(k, _)| k.to_string())
         .collect::<Vec<_>>();
@@ -197,3 +197,67 @@ pub fn compute_pipeline_diff(
         .with_modified_output_connectors(modified_output_connectors)
         .with_removed_output_connectors(removed_output_connectors))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::compute_pipeline_diff;
+    use feldera_types::config::PipelineConfig;
+    use serde_json::{Map, Value, json};
+
+    fn input_endpoint(mut fields: Map<String, Value>) -> Value {
+        let mut endpoint = Map::from_iter([
+            ("stream".to_string(), json!("t1")),
+            ("transport".to_string(), json!({"name": "empty_input"})),
+        ]);
+        endpoint.append(&mut fields);
+        Value::Object(endpoint)
+    }
+
+    fn pipeline_config(input: Value) -> PipelineConfig {
+        serde_json::from_value(json!({
+            "name": "test",
+            "workers": 1,
+            "inputs": {
+                "t1.connector": input
+            }
+        }))
+        .unwrap()
+    }
+
+    #[test]
+    fn input_flow_control_changes_do_not_modify_connector() {
+        let old_config = pipeline_config(input_endpoint(Map::new()));
+        let new_config = pipeline_config(input_endpoint(Map::from_iter([
+            ("max_queued_records".to_string(), json!(5000)),
+            ("max_queued_bytes".to_string(), json!(6000)),
+            ("max_batch_size".to_string(), json!(7000)),
+            ("max_worker_batch_size".to_string(), json!(8000)),
+            ("paused".to_string(), json!(true)),
+        ])));
+
+        let diff = compute_pipeline_diff(&old_config, &new_config).unwrap();
+
+        assert!(diff.modified_input_connectors().is_empty());
+    }
+
+    #[test]
+    fn input_transport_changes_still_modify_connector() {
+        let old_config = pipeline_config(input_endpoint(Map::new()));
+        let new_config = pipeline_config(input_endpoint(Map::from_iter([(
+            "transport".to_string(),
+            json!({
+                "name": "datagen",
+                "config": {
+                    "plan": [{"limit": 1}]
+                }
+            }),
+        )])));
+
+        let diff = compute_pipeline_diff(&old_config, &new_config).unwrap();
+
+        assert_eq!(
+            diff.modified_input_connectors(),
+            &vec!["t1.connector".to_string()]
+        );
+    }
+}
diff --git a/crates/feldera-types/src/config.rs b/crates/feldera-types/src/config.rs
@@ -1723,6 +1723,33 @@ impl ConnectorConfig {
         a == b
     }
 
+    /// Compare two input connector configs modulo fields that only affect
+    /// runtime flow control and do not invalidate checkpointed connector state.
+    pub fn equal_for_input_checkpoint_replay(&self, other: &Self) -> bool {
+        let mut a = self.clone();
+        let mut b = other.clone();
+        a.normalize_for_input_checkpoint_replay();
+        b.normalize_for_input_checkpoint_replay();
+        a == b
+    }
+
+    fn normalize_for_input_checkpoint_replay(&mut self) {
+        self.paused = false;
+        self.max_batch_size = None;
+        self.max_worker_batch_size = None;
+        self.max_queued_records = default_max_queued_records();
+        self.max_queued_bytes = None;
+    }
+
+    /// Adopt input connector settings that are safe to change while replaying
+    /// checkpointed connector state.
+    pub fn apply_input_checkpoint_replay_config_from(&mut self, other: &Self) {
+        self.max_batch_size = other.max_batch_size;
+        self.max_worker_batch_size = other.max_worker_batch_size;
+        self.max_queued_records = other.max_queued_records;
+        self.max_queued_bytes = other.max_queued_bytes;
+    }
+
     /// Returns `max_queued_records` or, if it is not set, the default.
     pub fn max_queued_records(&self) -> u64 {
         self.max_queued_records

diff --git a/python/tests/platform/test_bootstrapping.py b/python/tests/platform/test_bootstrapping.py
@@ -708,3 +708,66 @@ def gen_sql(connectors):
 
     pipeline.checkpoint(True)
     pipeline.stop(force=True)
+
+
+@enterprise_only
+@single_host_only
+@gen_pipeline_name
+def test_bootstrap_input_flow_control_changes(pipeline_name):
+    """
+    Enterprise: input connector flow-control changes should not require bootstrapping.
+    """
+
+    def gen_sql(max_queued_records):
+        connectors = json.dumps(
+            [
+                {
+                    "name": "datagen_connector",
+                    "transport": {
+                        "name": "datagen",
+                        "config": {"plan": [{"limit": 10}]},
+                    },
+                    "max_queued_records": max_queued_records,
+                    "max_queued_bytes": max_queued_records * 100,
+                    "max_batch_size": max_queued_records + 1,
+                    "max_worker_batch_size": max_queued_records + 2,
+                }
+            ]
+        )
+        return f"""CREATE TABLE t1(x int)
+with (
+    'materialized' = 'true',
+    'connectors' = '{connectors}'
+);
+CREATE MATERIALIZED VIEW v1 AS SELECT COUNT(*) AS c FROM t1;
+"""
+
+    pipeline = PipelineBuilder(
+        TEST_CLIENT,
+        pipeline_name,
+        sql=gen_sql(100),
+        runtime_config=RuntimeConfig(
+            workers=FELDERA_TEST_NUM_WORKERS,
+            hosts=FELDERA_TEST_NUM_HOSTS,
+            fault_tolerance_model=None,
+        ),
+    ).create_or_replace()
+
+    pipeline.start()
+    assert pipeline.status() == PipelineStatus.RUNNING
+
+    pipeline.wait_for_completion(timeout_s=300)
+    result = list(pipeline.query("SELECT * FROM v1;"))
+    assert result == [{"c": 10}]
+
+    pipeline.checkpoint(True)
+    pipeline.stop(force=True)
+
+    pipeline.modify(sql=gen_sql(200))
+    pipeline.start(bootstrap_policy=BootstrapPolicy.REJECT, timeout_s=300)
+    assert pipeline.status() == PipelineStatus.RUNNING
+
+    result = list(pipeline.query("SELECT * FROM v1;"))
+    assert result == [{"c": 10}]
+
+    pipeline.stop(force=True)