feldera · Dnreikronos · Jun 22, 2026 · Jun 22, 2026 · Jun 23, 2026 · Jun 24, 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,50 @@ 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<Option<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()) else {
+            return Ok(None);
+        };
+        factory.create(config)
+    }
+}
+
 #[doc(hidden)]
 pub trait IntegratedInputEndpoint: InputEndpoint {
     fn open(
@@ -1094,6 +1138,54 @@ 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<Option<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()) else {
+            return Ok(None);
+        };
+        factory.create(config, endpoint_name, fault_tolerant)
+    }
+}
+
 /// An [UnboundedReceiver] wrapper for [InputReaderCommand] for fault-tolerant connectors.
 ///
 /// A fault-tolerant connector wants to receive, in order:

diff --git a/crates/adapters/src/adhoc/table.rs b/crates/adapters/src/adhoc/table.rs
@@ -293,7 +293,7 @@ impl DataSink for AdHocTableSink {
         let config = InputEndpointConfig::new(
             self.name.to_string(),
             ConnectorConfig::new(
-                TransportConfig::AdHocInput(config),
+                TransportConfig::new(TransportConfig::ADHOC_INPUT, config),
                 Some(FormatConfig {
                     name: Cow::from("parquet"),
                     config: serde_json::Value::Null,

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
@@ -35,7 +35,6 @@ use crate::server::metrics::{HistogramDiv, LabelStack, MetricsFormatter, Metrics
 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::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};
@@ -4841,10 +4842,7 @@ impl ControllerInit {
                     .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(_)
-                        )
+                        config.connector_config.transport.name() != TransportConfig::CLOCK
                     })
                     .collect()
             } else {
@@ -6268,12 +6266,23 @@ 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 factory = self
+            .catalog
+            .input_transport_registry()
+            .lock()
+            .unwrap()
+            .get(transport_config.name());
+        let endpoint = match factory {
+            Some(factory) => factory
+                .create(&transport_config)
+                .map_err(|e| ControllerError::input_transport_error(endpoint_name, true, e))?,
+            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`.
@@ -6400,17 +6409,19 @@ impl ControllerInner {
                     &resolved_connector_config.transport,
                     &resolved_connector_config.format,
                 ) {
-                    (TransportConfig::Datagen(_), None) => FormatConfig {
-                        name: Cow::from("json"),
-                        config: serde_json::to_value(JsonParserConfig {
-                            update_format: JsonUpdateFormat::Raw,
-                            json_flavor: JsonFlavor::Datagen,
-                            array: true,
-                            lines: JsonLines::Multiple,
-                        })
-                        .unwrap(),
-                    },
-                    (TransportConfig::Datagen(_), Some(_)) => {
+                    (transport, None) if transport.name() == TransportConfig::DATAGEN => {
+                        FormatConfig {
+                            name: Cow::from("json"),
+                            config: serde_json::to_value(JsonParserConfig {
+                                update_format: JsonUpdateFormat::Raw,
+                                json_flavor: JsonFlavor::Datagen,
+                                array: true,
+                                lines: JsonLines::Multiple,
+                            })
+                            .unwrap(),
+                        }
+                    }
+                    (transport, Some(_)) if transport.name() == TransportConfig::DATAGEN => {
                         return Err(ControllerError::input_format_not_supported(
                             endpoint_name,
                             "datagen endpoints do not support custom formats: remove the 'format' section from connector specification",
@@ -6588,13 +6599,27 @@ 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 factory = self
+            .catalog
+            .output_transport_registry()
+            .lock()
+            .unwrap()
+            .get(transport_config.name());
+        let endpoint = match factory {
+            Some(factory) => factory
+                .create(
+                    &transport_config,
+                    endpoint_name,
+                    self.fault_tolerance == Some(FtModel::ExactlyOnce),
+                )
+                .map_err(|e| ControllerError::output_transport_error(endpoint_name, true, e))?,
+            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`.

diff --git a/crates/adapters/src/format/avro/output.rs b/crates/adapters/src/format/avro/output.rs
@@ -92,19 +92,25 @@ impl OutputFormat for AvroOutputFormat {
             )
         })?;
 
-        if matches!(
-            config.transport,
-            feldera_types::config::TransportConfig::RedisOutput(_)
-        ) {
+        if config.transport.name() == TransportConfig::REDIS_OUTPUT {
             return Err(ControllerError::invalid_encoder_configuration(
                 endpoint_name,
                 "'avro' format not yet supported with Redis connector",
             ));
         }
 
-        let topic = match &config.transport {
-            TransportConfig::KafkaOutput(kafka_config) => Some(kafka_config.topic.clone()),
-            _ => None,
+        let topic = if config.transport.name() == TransportConfig::KAFKA_OUTPUT {
+            let kafka_config: feldera_types::transport::kafka::KafkaOutputConfig =
+                config.transport.deserialize_config().map_err(|e| {
+                    ControllerError::encoder_config_parse_error(
+                        endpoint_name,
+                        &e,
+                        &serde_json::to_string(config).unwrap_or_default(),
+                    )
+                })?;
+            Some(kafka_config.topic)
+        } else {
+            None
         };
 
         if avro_config.threads == 0 {

diff --git a/crates/adapters/src/format/csv.rs b/crates/adapters/src/format/csv.rs
@@ -346,10 +346,7 @@ impl OutputFormat for CsvOutputFormat {
             )
         })?;
 
-        if matches!(
-            config.transport,
-            feldera_types::config::TransportConfig::RedisOutput(_)
-        ) {
+        if config.transport.name() == feldera_types::config::TransportConfig::REDIS_OUTPUT {
             return Err(ControllerError::invalid_encoder_configuration(
                 endpoint_name,
                 "'csv' format not yet supported with Redis connector",

diff --git a/crates/adapters/src/format/json/output.rs b/crates/adapters/src/format/json/output.rs
@@ -74,7 +74,7 @@ impl OutputFormat for JsonOutputFormat {
             )
         })?;
 
-        if matches!(&config.transport, TransportConfig::RedisOutput(_)) {
+        if config.transport.name() == TransportConfig::REDIS_OUTPUT {
             json_config.update_format = JsonUpdateFormat::Redis;
         };
 
@@ -88,9 +88,18 @@ impl OutputFormat for JsonOutputFormat {
             json_config.buffer_size_records = 1;
         }
 
-        let key_separator = match &config.transport {
-            TransportConfig::RedisOutput(config) => Some(config.key_separator.clone()),
-            _ => None,
+        let key_separator = if config.transport.name() == TransportConfig::REDIS_OUTPUT {
+            let redis_config: feldera_types::transport::redis::RedisOutputConfig =
+                config.transport.deserialize_config().map_err(|e| {
+                    ControllerError::encoder_config_parse_error(
+                        endpoint_name,
+                        &e,
+                        &serde_json::to_string(config).unwrap_or_default(),
+                    )
+                })?;
+            Some(redis_config.key_separator)
+        } else {
+            None
         };
 
         Ok(Box::new(JsonEncoder::new(