feldera · Dnreikronos · Jun 22, 2026 · Jun 22, 2026 · Jun 24, 2026 · Jul 3, 2026
diff --git a/crates/adapterlib/src/catalog.rs b/crates/adapterlib/src/catalog.rs
@@ -31,6 +31,7 @@ use crate::errors::controller::ControllerError;
 use crate::format::InputBuffer;
 use crate::postprocess::PostprocessorRegistry;
 use crate::preprocess::PreprocessorRegistry;
+use crate::transport::{InputTransportRegistry, OutputTransportRegistry};
 
 /// Descriptor that specifies the format in which records are received
 /// or into which they should be encoded before sending.
@@ -1166,6 +1167,12 @@ pub trait CircuitCatalog: Send + Sync {
 
     /// The registry used to insert new user-defined postprocessors
     fn postprocessor_registry(&self) -> Arc<Mutex<PostprocessorRegistry>>;
+
+    /// The registry used to look up input transport endpoint factories.
+    fn input_transport_registry(&self) -> Arc<Mutex<InputTransportRegistry>>;
+
+    /// The registry used to look up output transport endpoint factories.
+    fn output_transport_registry(&self) -> Arc<Mutex<OutputTransportRegistry>>;
 }
 
 #[doc(hidden)]

diff --git a/crates/adapterlib/src/transport.rs b/crates/adapterlib/src/transport.rs
@@ -2,14 +2,14 @@ use anyhow::{Error as AnyError, Result as AnyResult};
 use chrono::{DateTime, Utc};
 use dyn_clone::DynClone;
 use feldera_types::adapter_stats::ConnectorHealth;
-use feldera_types::config::FtModel;
+use feldera_types::config::{FtModel, TransportConfig};
 use feldera_types::coordination::Completion;
 use feldera_types::program_schema::Relation;
 use rmpv::{Value as RmpValue, ext::Error as RmpDecodeError};
 use serde::Deserialize;
 use serde::de::DeserializeOwned;
 use serde_json::Value as JsonValue;
-use std::collections::VecDeque;
+use std::collections::{BTreeMap, VecDeque};
 use std::fmt::Display;
 use std::marker::PhantomData;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -75,6 +75,47 @@ pub trait TransportInputEndpoint: InputEndpoint {
     ) -> AnyResult<Box<dyn InputReader>>;
 }
 
+/// Factory for creating input transport endpoints from transport configuration.
+pub trait InputTransportEndpointFactory: Send + Sync {
+    fn create(&self, config: &TransportConfig) -> AnyResult<Box<dyn TransportInputEndpoint>>;
+}
+
+/// Registry of input transport endpoint factories keyed by transport name.
+#[derive(Default)]
+pub struct InputTransportRegistry {
+    registered: BTreeMap<String, Arc<dyn InputTransportEndpointFactory>>,
+}
+
+impl InputTransportRegistry {
+    pub fn new() -> Self {
+        Self {
+            registered: BTreeMap::new(),
+        }
+    }
+
+    pub fn register(
+        &mut self,
+        name: impl Into<String>,
+        factory: Box<dyn InputTransportEndpointFactory>,
+    ) {
+        self.registered.insert(name.into(), Arc::from(factory));
+    }
+
+    pub fn get(&self, name: &str) -> Option<Arc<dyn InputTransportEndpointFactory>> {
+        self.registered.get(name).cloned()
+    }
+
+    pub fn create_endpoint(
+        &self,
+        config: &TransportConfig,
+    ) -> AnyResult<Option<Box<dyn TransportInputEndpoint>>> {
+        let Some(factory) = self.get(config.name().as_str()) else {
+            return Ok(None);
+        };
+        factory.create(config).map(Some)
+    }
+}
+
 #[doc(hidden)]
 pub trait IntegratedInputEndpoint: InputEndpoint {
     fn open(
@@ -1094,6 +1135,56 @@ pub trait OutputEndpoint: Send {
     }
 }
 
+/// Factory for creating output transport endpoints from transport configuration.
+pub trait OutputTransportEndpointFactory: Send + Sync {
+    fn create(
+        &self,
+        config: &TransportConfig,
+        endpoint_name: &str,
+        fault_tolerant: bool,
+    ) -> AnyResult<Box<dyn OutputEndpoint>>;
+}
+
+/// Registry of output transport endpoint factories keyed by transport name.
+#[derive(Default)]
+pub struct OutputTransportRegistry {
+    registered: BTreeMap<String, Arc<dyn OutputTransportEndpointFactory>>,
+}
+
+impl OutputTransportRegistry {
+    pub fn new() -> Self {
+        Self {
+            registered: BTreeMap::new(),
+        }
+    }
+
+    pub fn register(
+        &mut self,
+        name: impl Into<String>,
+        factory: Box<dyn OutputTransportEndpointFactory>,
+    ) {
+        self.registered.insert(name.into(), Arc::from(factory));
+    }
+
+    pub fn get(&self, name: &str) -> Option<Arc<dyn OutputTransportEndpointFactory>> {
+        self.registered.get(name).cloned()
+    }
+
+    pub fn create_endpoint(
+        &self,
+        config: &TransportConfig,
+        endpoint_name: &str,
+        fault_tolerant: bool,
+    ) -> AnyResult<Option<Box<dyn OutputEndpoint>>> {
+        let Some(factory) = self.get(config.name().as_str()) else {
+            return Ok(None);
+        };
+        factory
+            .create(config, endpoint_name, fault_tolerant)
+            .map(Some)
+    }
+}
+
 /// An [UnboundedReceiver] wrapper for [InputReaderCommand] for fault-tolerant connectors.
 ///
 /// A fault-tolerant connector wants to receive, in order:

diff --git a/crates/adapters/src/catalog.rs b/crates/adapters/src/catalog.rs
@@ -1,6 +1,8 @@
 use feldera_adapterlib::{
-    errors::controller::ControllerError, postprocess::PostprocessorRegistry,
+    errors::controller::ControllerError,
+    postprocess::PostprocessorRegistry,
     preprocess::PreprocessorRegistry,
+    transport::{InputTransportRegistry, OutputTransportRegistry},
 };
 use feldera_types::program_schema::SqlIdentifier;
 use std::{
@@ -14,6 +16,8 @@ pub use feldera_adapterlib::catalog::*;
 pub struct Catalog {
     input_collection_handles: BTreeMap<SqlIdentifier, InputCollectionHandle>,
     output_batch_handles: BTreeMap<SqlIdentifier, OutputCollectionHandles>,
+    input_transport_registry: Arc<Mutex<InputTransportRegistry>>,
+    output_transport_registry: Arc<Mutex<OutputTransportRegistry>>,
     preprocessor_registry: Arc<Mutex<PreprocessorRegistry>>,
     postprocessor_registry: Arc<Mutex<PostprocessorRegistry>>,
 }
@@ -29,6 +33,12 @@ impl Catalog {
         Self {
             input_collection_handles: BTreeMap::new(),
             output_batch_handles: BTreeMap::new(),
+            input_transport_registry: Arc::new(Mutex::new(
+                crate::transport::builtin_input_transport_registry(),
+            )),
+            output_transport_registry: Arc::new(Mutex::new(
+                crate::transport::builtin_output_transport_registry(),
+            )),
             preprocessor_registry: Arc::new(Mutex::new(PreprocessorRegistry::new())),
             postprocessor_registry: Arc::new(Mutex::new(PostprocessorRegistry::new())),
         }
@@ -128,4 +138,12 @@ impl CircuitCatalog for Catalog {
     fn postprocessor_registry(&self) -> Arc<Mutex<PostprocessorRegistry>> {
         self.postprocessor_registry.clone()
     }
+
+    fn input_transport_registry(&self) -> Arc<Mutex<InputTransportRegistry>> {
+        self.input_transport_registry.clone()
+    }
+
+    fn output_transport_registry(&self) -> Arc<Mutex<OutputTransportRegistry>> {
+        self.output_transport_registry.clone()
+    }
 }
diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
@@ -33,9 +33,8 @@ use crate::controller::sync::{
 use crate::panic::N_PANICS;
 use crate::server::metrics::{HistogramDiv, LabelStack, MetricsFormatter, MetricsWriter, Value};
 use crate::server::{InitializationState, ServerState};
-use crate::transport::Step;
 use crate::transport::clock::now_endpoint_config;
-use crate::transport::{input_transport_config_to_endpoint, output_transport_config_to_endpoint};
+use crate::transport::{Step, input_transport_uses_registry, output_transport_uses_registry};
 use crate::util::{LongOperationWarning, run_on_thread_pool};
 use crate::{
     CircuitCatalog, Encoder, InputConsumer, OutputConsumer, OutputEndpoint, ParseError,
@@ -94,7 +93,9 @@ use feldera_types::coordination::{
 use feldera_types::format::json::JsonLines;
 use feldera_types::pipeline_diff::PipelineDiff;
 use feldera_types::runtime_status::BootstrapPolicy;
-use feldera_types::secret_resolver::resolve_secret_references_in_connector_config;
+use feldera_types::secret_resolver::{
+    resolve_secret_references_in_connector_config, resolve_secret_references_via_json,
+};
 use feldera_types::suspend::{PermanentSuspendError, SuspendError, TemporarySuspendError};
 use feldera_types::time_series::SampleStatistics;
 use feldera_types::transaction::{StartTransactionResponse, TransactionId};
@@ -6268,12 +6269,32 @@ impl ControllerInner {
         endpoint_config: &InputEndpointConfig,
         resume_info: Option<(JsonValue, CheckpointInputEndpointMetrics)>,
     ) -> Result<EndpointId, ControllerError> {
-        let endpoint = input_transport_config_to_endpoint(
-            &endpoint_config.connector_config.transport,
-            endpoint_name,
+        let transport_config = resolve_secret_references_via_json(
             &self.secrets_dir,
+            &endpoint_config.connector_config.transport,
         )
         .map_err(|e| ControllerError::input_transport_error(endpoint_name, true, e))?;
+        let transport_name = transport_config.name();
+        let factory = self
+            .catalog
+            .input_transport_registry()
+            .lock()
+            .unwrap()
+            .get(&transport_name);
+        let endpoint = match factory {
+            Some(factory) => Some(
+                factory
+                    .create(&transport_config)
+                    .map_err(|e| ControllerError::input_transport_error(endpoint_name, true, e))?,
+            ),
+            None if input_transport_uses_registry(&transport_config) => {
+                return Err(ControllerError::unknown_input_transport(
+                    endpoint_name,
+                    &transport_name,
+                ));
+            }
+            None => None,
+        };
 
         // If `endpoint` is `None`, it means that the endpoint config specifies an integrated
         // input connector.  Such endpoints are instantiated inside `add_input_endpoint`.
@@ -6588,13 +6609,36 @@ impl ControllerInner {
         endpoint_config: &OutputEndpointConfig,
         initial_statistics: Option<&CheckpointOutputEndpointMetrics>,
     ) -> Result<EndpointId, ControllerError> {
-        let endpoint = output_transport_config_to_endpoint(
-            &endpoint_config.connector_config.transport,
-            endpoint_name,
-            self.fault_tolerance == Some(FtModel::ExactlyOnce),
+        let transport_config = resolve_secret_references_via_json(
             &self.secrets_dir,
+            &endpoint_config.connector_config.transport,
         )
         .map_err(|e| ControllerError::output_transport_error(endpoint_name, true, e))?;
+        let transport_name = transport_config.name();
+        let factory = self
+            .catalog
+            .output_transport_registry()
+            .lock()
+            .unwrap()
+            .get(&transport_name);
+        let endpoint = match factory {
+            Some(factory) => Some(
+                factory
+                    .create(
+                        &transport_config,
+                        endpoint_name,
+                        self.fault_tolerance == Some(FtModel::ExactlyOnce),
+                    )
+                    .map_err(|e| ControllerError::output_transport_error(endpoint_name, true, e))?,
+            ),
+            None if output_transport_uses_registry(&transport_config) => {
+                return Err(ControllerError::unknown_output_transport(
+                    endpoint_name,
+                    &transport_name,
+                ));
+            }
+            None => None,
+        };
 
         // If `endpoint` is `None`, it means that the endpoint config specifies an integrated
         // output connector.  Such endpoints are instantiated inside `add_output_endpoint`.