From 70040dad0c5733adb9b055d485880d7ffae98782 Mon Sep 17 00:00:00 2001 From: Dnreikronos Date: Mon, 22 Jun 2026 17:33:10 -0300 Subject: [PATCH 1/4] Add connector transport registries --- crates/adapterlib/src/catalog.rs | 7 + crates/adapterlib/src/transport.rs | 96 +++++- crates/adapters/src/catalog.rs | 20 +- crates/adapters/src/controller.rs | 44 ++- crates/adapters/src/integrated.rs | 480 +++++++++++++++++++++++++---- crates/adapters/src/transport.rs | 454 +++++++++++++++++++++++---- 6 files changed, 969 insertions(+), 132 deletions(-) diff --git a/crates/adapterlib/src/catalog.rs b/crates/adapterlib/src/catalog.rs index 6260edc6c33..a3fb5d031ef 100644 --- 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>; + + /// The registry used to look up input transport endpoint factories. + fn input_transport_registry(&self) -> Arc>; + + /// The registry used to look up output transport endpoint factories. + fn output_transport_registry(&self) -> Arc>; } #[doc(hidden)] diff --git a/crates/adapterlib/src/transport.rs b/crates/adapterlib/src/transport.rs index a5b3c81cc1b..7225408b055 100644 --- 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>; } +/// Factory for creating input transport endpoints from transport configuration. +pub trait InputTransportEndpointFactory: Send + Sync { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>>; +} + +/// Registry of input transport endpoint factories keyed by transport name. +#[derive(Default)] +pub struct InputTransportRegistry { + registered: BTreeMap<&'static str, Arc>, +} + +impl InputTransportRegistry { + pub fn new() -> Self { + Self { + registered: BTreeMap::new(), + } + } + + pub fn register( + &mut self, + name: &'static str, + factory: Box, + ) { + self.registered.insert(name, Arc::from(factory)); + } + + pub fn get(&self, name: &str) -> Option> { + self.registered.get(name).cloned() + } + + pub fn create_endpoint( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + 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>>; +} + +/// Registry of output transport endpoint factories keyed by transport name. +#[derive(Default)] +pub struct OutputTransportRegistry { + registered: BTreeMap<&'static str, Arc>, +} + +impl OutputTransportRegistry { + pub fn new() -> Self { + Self { + registered: BTreeMap::new(), + } + } + + pub fn register( + &mut self, + name: &'static str, + factory: Box, + ) { + self.registered.insert(name, Arc::from(factory)); + } + + pub fn get(&self, name: &str) -> Option> { + self.registered.get(name).cloned() + } + + pub fn create_endpoint( + &self, + config: &TransportConfig, + endpoint_name: &str, + fault_tolerant: bool, + ) -> AnyResult>> { + 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/catalog.rs b/crates/adapters/src/catalog.rs index 2b367b14a1a..5fdb70092b8 100644 --- 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, output_batch_handles: BTreeMap, + input_transport_registry: Arc>, + output_transport_registry: Arc>, preprocessor_registry: Arc>, postprocessor_registry: Arc>, } @@ -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> { self.postprocessor_registry.clone() } + + fn input_transport_registry(&self) -> Arc> { + self.input_transport_registry.clone() + } + + fn output_transport_registry(&self) -> Arc> { + self.output_transport_registry.clone() + } } diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index e90ee69795d..8864a11540f 100644 --- 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}; @@ -6268,12 +6269,23 @@ impl ControllerInner { endpoint_config: &InputEndpointConfig, resume_info: Option<(JsonValue, CheckpointInputEndpointMetrics)>, ) -> Result { - 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`. @@ -6588,13 +6600,27 @@ impl ControllerInner { endpoint_config: &OutputEndpointConfig, initial_statistics: Option<&CheckpointOutputEndpointMetrics>, ) -> Result { - 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/integrated.rs b/crates/adapters/src/integrated.rs index a36df0f8d3d..baea053d1d3 100644 --- a/crates/adapters/src/integrated.rs +++ b/crates/adapters/src/integrated.rs @@ -4,7 +4,10 @@ use crate::{ControllerError, Encoder, InputConsumer, OutputEndpoint}; use datafusion::execution::runtime_env::RuntimeEnv; use feldera_types::config::{ConnectorConfig, PipelineConfig, TransportConfig}; use feldera_types::program_schema::Relation; -use std::sync::{Arc, Weak}; +use std::{ + collections::BTreeMap, + sync::{Arc, Weak}, +}; #[cfg(feature = "with-deltalake")] pub mod delta_table; @@ -40,6 +43,354 @@ where } } +/// Factory for creating integrated output endpoints from connector configuration. +pub trait IntegratedOutputEndpointFactory: Send + Sync { + #[allow(clippy::too_many_arguments)] + fn create( + &self, + endpoint_id: EndpointId, + endpoint_name: &str, + connector_config: &ConnectorConfig, + key_schema: &Option, + schema: &Relation, + controller: Weak, + continue_previous_state: bool, + is_index: bool, + ) -> Result>, ControllerError>; +} + +/// Registry of integrated output endpoint factories keyed by transport name. +#[derive(Default)] +pub struct IntegratedOutputEndpointRegistry { + registered: BTreeMap<&'static str, Arc>, +} + +impl IntegratedOutputEndpointRegistry { + pub fn new() -> Self { + Self { + registered: BTreeMap::new(), + } + } + + pub fn register( + &mut self, + name: &'static str, + factory: Box, + ) { + self.registered.insert(name, Arc::from(factory)); + } + + pub fn get(&self, name: &str) -> Option> { + self.registered.get(name).cloned() + } + + #[allow(clippy::too_many_arguments)] + pub fn create_endpoint( + &self, + endpoint_id: EndpointId, + endpoint_name: &str, + connector_config: &ConnectorConfig, + key_schema: &Option, + schema: &Relation, + controller: Weak, + continue_previous_state: bool, + is_index: bool, + ) -> Result>, ControllerError> { + let Some(factory) = self.get(&connector_config.transport.name()) else { + return Ok(None); + }; + factory.create( + endpoint_id, + endpoint_name, + connector_config, + key_schema, + schema, + controller, + continue_previous_state, + is_index, + ) + } +} + +/// Factory for creating integrated input endpoints from connector configuration. +pub trait IntegratedInputEndpointFactory: Send + Sync { + fn create( + &self, + endpoint_name: &str, + config: &ConnectorConfig, + pipeline_config: &PipelineConfig, + runtime_env: Arc, + consumer: Box, + ) -> Result>, ControllerError>; +} + +/// Registry of integrated input endpoint factories keyed by transport name. +#[derive(Default)] +pub struct IntegratedInputEndpointRegistry { + registered: BTreeMap<&'static str, Arc>, +} + +impl IntegratedInputEndpointRegistry { + pub fn new() -> Self { + Self { + registered: BTreeMap::new(), + } + } + + pub fn register( + &mut self, + name: &'static str, + factory: Box, + ) { + self.registered.insert(name, Arc::from(factory)); + } + + pub fn get(&self, name: &str) -> Option> { + self.registered.get(name).cloned() + } + + pub fn create_endpoint( + &self, + endpoint_name: &str, + config: &ConnectorConfig, + pipeline_config: &PipelineConfig, + runtime_env: Arc, + consumer: Box, + ) -> Result>, ControllerError> { + let Some(factory) = self.get(&config.transport.name()) else { + return Ok(None); + }; + factory.create( + endpoint_name, + config, + pipeline_config, + runtime_env, + consumer, + ) + } +} + +pub fn builtin_integrated_output_endpoint_registry() -> IntegratedOutputEndpointRegistry { + let mut registry = IntegratedOutputEndpointRegistry::new(); + #[cfg(feature = "with-deltalake")] + registry.register("delta_table_output", Box::new(DeltaTableOutputFactory)); + registry.register("postgres_output", Box::new(PostgresOutputFactory)); + #[cfg(feature = "with-dynamodb")] + registry.register("dynamodb_output", Box::new(DynamoDBOutputFactory)); + registry +} + +pub fn builtin_integrated_input_endpoint_registry() -> IntegratedInputEndpointRegistry { + let mut registry = IntegratedInputEndpointRegistry::new(); + #[cfg(feature = "with-deltalake")] + registry.register("delta_table_input", Box::new(DeltaTableInputFactory)); + #[cfg(feature = "with-iceberg")] + registry.register("iceberg_input", Box::new(IcebergInputFactory)); + registry.register("postgres_input", Box::new(PostgresInputFactory)); + #[cfg(feature = "with-postgres-cdc")] + registry.register("postgres_cdc_input", Box::new(PostgresCdcInputFactory)); + registry +} + +#[cfg(feature = "with-deltalake")] +struct DeltaTableOutputFactory; + +#[cfg(feature = "with-deltalake")] +impl IntegratedOutputEndpointFactory for DeltaTableOutputFactory { + #[allow(clippy::too_many_arguments)] + fn create( + &self, + endpoint_id: EndpointId, + endpoint_name: &str, + connector_config: &ConnectorConfig, + key_schema: &Option, + schema: &Relation, + controller: Weak, + continue_previous_state: bool, + is_index: bool, + ) -> Result>, ControllerError> { + match &connector_config.transport { + TransportConfig::DeltaTableOutput(config) => { + Ok(Some(Box::new(delta_table::DeltaTableWriter::new( + endpoint_id, + endpoint_name, + config, + key_schema, + schema, + controller, + continue_previous_state, + is_index, + )?))) + } + _ => Ok(None), + } + } +} + +struct PostgresOutputFactory; + +impl IntegratedOutputEndpointFactory for PostgresOutputFactory { + #[allow(clippy::too_many_arguments)] + fn create( + &self, + endpoint_id: EndpointId, + endpoint_name: &str, + connector_config: &ConnectorConfig, + key_schema: &Option, + schema: &Relation, + controller: Weak, + _continue_previous_state: bool, + is_index: bool, + ) -> Result>, ControllerError> { + match &connector_config.transport { + TransportConfig::PostgresOutput(config) => { + Ok(Some(Box::new(PostgresOutputEndpoint::new( + endpoint_id, + endpoint_name, + config, + key_schema, + schema, + controller, + is_index, + )?))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-dynamodb")] +struct DynamoDBOutputFactory; + +#[cfg(feature = "with-dynamodb")] +impl IntegratedOutputEndpointFactory for DynamoDBOutputFactory { + #[allow(clippy::too_many_arguments)] + fn create( + &self, + endpoint_id: EndpointId, + endpoint_name: &str, + connector_config: &ConnectorConfig, + key_schema: &Option, + schema: &Relation, + controller: Weak, + _continue_previous_state: bool, + is_index: bool, + ) -> Result>, ControllerError> { + match &connector_config.transport { + TransportConfig::DynamoDBOutput(config) => { + Ok(Some(Box::new(DynamoDBOutputEndpoint::new( + endpoint_id, + endpoint_name, + config, + key_schema, + schema, + controller, + is_index, + )?))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-deltalake")] +struct DeltaTableInputFactory; + +#[cfg(feature = "with-deltalake")] +impl IntegratedInputEndpointFactory for DeltaTableInputFactory { + fn create( + &self, + endpoint_name: &str, + config: &ConnectorConfig, + pipeline_config: &PipelineConfig, + runtime_env: Arc, + consumer: Box, + ) -> Result>, ControllerError> { + match &config.transport { + TransportConfig::DeltaTableInput(config) => { + Ok(Some(Box::new(delta_table::DeltaTableInputEndpoint::new( + endpoint_name, + config, + pipeline_config, + runtime_env, + consumer, + )))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-iceberg")] +struct IcebergInputFactory; + +#[cfg(feature = "with-iceberg")] +impl IntegratedInputEndpointFactory for IcebergInputFactory { + fn create( + &self, + endpoint_name: &str, + config: &ConnectorConfig, + pipeline_config: &PipelineConfig, + runtime_env: Arc, + consumer: Box, + ) -> Result>, ControllerError> { + match &config.transport { + TransportConfig::IcebergInput(config) => { + Ok(Some(Box::new(feldera_iceberg::IcebergInputEndpoint::new( + endpoint_name, + config, + pipeline_config, + runtime_env, + consumer, + )))) + } + _ => Ok(None), + } + } +} + +struct PostgresInputFactory; + +impl IntegratedInputEndpointFactory for PostgresInputFactory { + fn create( + &self, + endpoint_name: &str, + config: &ConnectorConfig, + _pipeline_config: &PipelineConfig, + _runtime_env: Arc, + consumer: Box, + ) -> Result>, ControllerError> { + match &config.transport { + TransportConfig::PostgresInput(config) => Ok(Some(Box::new( + PostgresInputEndpoint::new(endpoint_name, config, consumer), + ))), + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-postgres-cdc")] +struct PostgresCdcInputFactory; + +#[cfg(feature = "with-postgres-cdc")] +impl IntegratedInputEndpointFactory for PostgresCdcInputFactory { + fn create( + &self, + endpoint_name: &str, + config: &ConnectorConfig, + _pipeline_config: &PipelineConfig, + _runtime_env: Arc, + consumer: Box, + ) -> Result>, ControllerError> { + match &config.transport { + TransportConfig::PostgresCdcInput(config) => Ok(Some(Box::new( + PostgresCdcInputEndpoint::new(endpoint_name, config, consumer), + ))), + _ => Ok(None), + } + } +} + /// Create an instance of an integrated output endpoint given its config /// and output relation schema. #[allow(unused, clippy::too_many_arguments)] @@ -53,44 +404,49 @@ pub fn create_integrated_output_endpoint( continue_previous_state: bool, is_index: bool, ) -> Result, ControllerError> { - let ep: Box = match &connector_config.transport { - #[cfg(feature = "with-deltalake")] - TransportConfig::DeltaTableOutput(config) => Box::new(delta_table::DeltaTableWriter::new( + let registry = builtin_integrated_output_endpoint_registry(); + create_integrated_output_endpoint_with_registry( + ®istry, + endpoint_id, + endpoint_name, + connector_config, + key_schema, + schema, + controller, + continue_previous_state, + is_index, + ) +} + +#[allow(unused, clippy::too_many_arguments)] +pub fn create_integrated_output_endpoint_with_registry( + registry: &IntegratedOutputEndpointRegistry, + endpoint_id: EndpointId, + endpoint_name: &str, + connector_config: &ConnectorConfig, + key_schema: &Option, + schema: &Relation, + controller: Weak, + continue_previous_state: bool, + is_index: bool, +) -> Result, ControllerError> { + let ep = registry + .create_endpoint( endpoint_id, endpoint_name, - config, + connector_config, key_schema, schema, controller, continue_previous_state, is_index, - )?), - TransportConfig::PostgresOutput(config) => Box::new(PostgresOutputEndpoint::new( - endpoint_id, - endpoint_name, - config, - key_schema, - schema, - controller, - is_index, - )?), - #[cfg(feature = "with-dynamodb")] - TransportConfig::DynamoDBOutput(config) => Box::new(DynamoDBOutputEndpoint::new( - endpoint_id, - endpoint_name, - config, - key_schema, - schema, - controller, - is_index, - )?), - transport => { - return Err(ControllerError::unknown_output_transport( + )? + .ok_or_else(|| { + ControllerError::unknown_output_transport( endpoint_name, - &transport.name(), - )); - } - }; + &connector_config.transport.name(), + ) + })?; if connector_config.format.is_some() { return Err(ControllerError::invalid_parser_configuration( @@ -113,43 +469,37 @@ pub fn create_integrated_input_endpoint( runtime_env: Arc, consumer: Box, ) -> Result, ControllerError> { - let ep: Box = match &config.transport { - #[cfg(feature = "with-deltalake")] - TransportConfig::DeltaTableInput(config) => { - Box::new(delta_table::DeltaTableInputEndpoint::new( - endpoint_name, - config, - pipeline_config, - runtime_env, - consumer, - )) - } - #[cfg(feature = "with-iceberg")] - TransportConfig::IcebergInput(config) => { - Box::new(feldera_iceberg::IcebergInputEndpoint::new( - endpoint_name, - config, - pipeline_config, - runtime_env, - consumer, - )) - } - TransportConfig::PostgresInput(config) => { - Box::new(PostgresInputEndpoint::new(endpoint_name, config, consumer)) - } - #[cfg(feature = "with-postgres-cdc")] - TransportConfig::PostgresCdcInput(config) => Box::new(PostgresCdcInputEndpoint::new( + let registry = builtin_integrated_input_endpoint_registry(); + create_integrated_input_endpoint_with_registry( + ®istry, + endpoint_name, + config, + pipeline_config, + runtime_env, + consumer, + ) +} + +#[allow(unused_variables)] +pub fn create_integrated_input_endpoint_with_registry( + registry: &IntegratedInputEndpointRegistry, + endpoint_name: &str, + config: &ConnectorConfig, + pipeline_config: &PipelineConfig, + runtime_env: Arc, + consumer: Box, +) -> Result, ControllerError> { + let ep = registry + .create_endpoint( endpoint_name, config, + pipeline_config, + runtime_env, consumer, - )), - transport => { - return Err(ControllerError::unknown_input_transport( - endpoint_name, - &transport.name(), - )); - } - }; + )? + .ok_or_else(|| { + ControllerError::unknown_input_transport(endpoint_name, &config.transport.name()) + })?; if config.format.is_some() { return Err(ControllerError::invalid_parser_configuration( diff --git a/crates/adapters/src/transport.rs b/crates/adapters/src/transport.rs index f634e86c0e4..d5ad42d7b21 100644 --- a/crates/adapters/src/transport.rs +++ b/crates/adapters/src/transport.rs @@ -80,6 +80,308 @@ use feldera_datagen::GeneratorEndpoint; pub use feldera_adapterlib::transport::*; +pub fn builtin_input_transport_registry() -> InputTransportRegistry { + let mut registry = InputTransportRegistry::new(); + registry.register("file_input", Box::new(FileInputFactory)); + #[cfg(feature = "with-kafka")] + registry.register("kafka_input", Box::new(KafkaInputFactory)); + #[cfg(feature = "with-nats")] + registry.register("nats_input", Box::new(NatsInputFactory)); + #[cfg(feature = "with-pubsub")] + registry.register("pub_sub_input", Box::new(PubSubInputFactory)); + registry.register("url_input", Box::new(UrlInputFactory)); + registry.register("s3_input", Box::new(S3InputFactory)); + registry.register("datagen", Box::new(DatagenInputFactory)); + #[cfg(feature = "with-nexmark")] + registry.register("nexmark", Box::new(NexmarkInputFactory)); + registry.register("http_input", Box::new(HttpInputFactory)); + registry.register("adhoc_input", Box::new(AdHocInputFactory)); + registry.register("clock", Box::new(ClockInputFactory)); + registry.register("empty_input", Box::new(EmptyInputFactory)); + registry +} + +pub fn builtin_output_transport_registry() -> OutputTransportRegistry { + let mut registry = OutputTransportRegistry::new(); + registry.register("file_output", Box::new(FileOutputFactory)); + #[cfg(feature = "with-kafka")] + registry.register("kafka_output", Box::new(KafkaOutputFactory)); + #[cfg(feature = "with-redis")] + registry.register("redis_output", Box::new(RedisOutputFactory)); + registry.register("null_output", Box::new(NullOutputFactory)); + registry +} + +struct FileInputFactory; + +impl InputTransportEndpointFactory for FileInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::FileInput(config) => { + Ok(Some(Box::new(FileInputEndpoint::new(config)))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-kafka")] +struct KafkaInputFactory; + +#[cfg(feature = "with-kafka")] +impl InputTransportEndpointFactory for KafkaInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::KafkaInput(config) => { + Ok(Some(Box::new(KafkaFtInputEndpoint::new(config)?))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-nats")] +struct NatsInputFactory; + +#[cfg(feature = "with-nats")] +impl InputTransportEndpointFactory for NatsInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::NatsInput(config) => { + Ok(Some(Box::new(NatsInputEndpoint::new(config)?))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-pubsub")] +struct PubSubInputFactory; + +#[cfg(feature = "with-pubsub")] +impl InputTransportEndpointFactory for PubSubInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::PubSubInput(config) => { + Ok(Some(Box::new(PubSubInputEndpoint::new(config.clone())?))) + } + _ => Ok(None), + } + } +} + +struct UrlInputFactory; + +impl InputTransportEndpointFactory for UrlInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::UrlInput(config) => Ok(Some(Box::new(UrlInputEndpoint::new(config)))), + _ => Ok(None), + } + } +} + +struct S3InputFactory; + +impl InputTransportEndpointFactory for S3InputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::S3Input(config) => Ok(Some(Box::new(S3InputEndpoint::new(config)?))), + _ => Ok(None), + } + } +} + +struct DatagenInputFactory; + +impl InputTransportEndpointFactory for DatagenInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::Datagen(config) => { + Ok(Some(Box::new(GeneratorEndpoint::new(config.clone())))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-nexmark")] +struct NexmarkInputFactory; + +#[cfg(feature = "with-nexmark")] +impl InputTransportEndpointFactory for NexmarkInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::Nexmark(config) => { + Ok(Some(Box::new(NexmarkEndpoint::new(config.clone())))) + } + _ => Ok(None), + } + } +} + +struct HttpInputFactory; + +impl InputTransportEndpointFactory for HttpInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::HttpInput(config) => { + Ok(Some(Box::new(HttpInputEndpoint::new(config)))) + } + _ => Ok(None), + } + } +} + +struct AdHocInputFactory; + +impl InputTransportEndpointFactory for AdHocInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::AdHocInput(config) => { + Ok(Some(Box::new(AdHocInputEndpoint::new(config)))) + } + _ => Ok(None), + } + } +} + +struct ClockInputFactory; + +impl InputTransportEndpointFactory for ClockInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::ClockInput(config) => Ok(Some(Box::new(ClockEndpoint::new(config)?))), + _ => Ok(None), + } + } +} + +struct EmptyInputFactory; + +impl InputTransportEndpointFactory for EmptyInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::EmptyInput => Ok(Some(Box::new(EmptyInputEndpoint))), + _ => Ok(None), + } + } +} + +struct FileOutputFactory; + +impl OutputTransportEndpointFactory for FileOutputFactory { + fn create( + &self, + config: &TransportConfig, + _endpoint_name: &str, + _fault_tolerant: bool, + ) -> AnyResult>> { + match config { + TransportConfig::FileOutput(config) => { + Ok(Some(Box::new(FileOutputEndpoint::new(config)?))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-kafka")] +struct KafkaOutputFactory; + +#[cfg(feature = "with-kafka")] +impl OutputTransportEndpointFactory for KafkaOutputFactory { + fn create( + &self, + config: &TransportConfig, + endpoint_name: &str, + fault_tolerant: bool, + ) -> AnyResult>> { + match config { + TransportConfig::KafkaOutput(config) => match fault_tolerant { + false => Ok(Some(Box::new(KafkaOutputEndpoint::new( + config, + endpoint_name, + )?))), + true => Ok(Some(Box::new(KafkaFtOutputEndpoint::new(config)?))), + }, + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-redis")] +struct RedisOutputFactory; + +#[cfg(feature = "with-redis")] +impl OutputTransportEndpointFactory for RedisOutputFactory { + fn create( + &self, + config: &TransportConfig, + _endpoint_name: &str, + _fault_tolerant: bool, + ) -> AnyResult>> { + match config { + TransportConfig::RedisOutput(config) => { + Ok(Some(Box::new(RedisOutputEndpoint::new(config)?))) + } + _ => Ok(None), + } + } +} + +struct NullOutputFactory; + +impl OutputTransportEndpointFactory for NullOutputFactory { + fn create( + &self, + config: &TransportConfig, + _endpoint_name: &str, + _fault_tolerant: bool, + ) -> AnyResult>> { + match config { + TransportConfig::NullOutput => Ok(Some(Box::new(NullOutputEndpoint))), + _ => Ok(None), + } + } +} + /// Creates an input transport endpoint instance using an input transport /// configuration, resolving secrets by reading `secrets_dir`. /// @@ -92,45 +394,7 @@ pub fn input_transport_config_to_endpoint( secrets_dir: &Path, ) -> AnyResult>> { let config = resolve_secret_references_via_json(secrets_dir, config)?; - let endpoint: Box = match config { - TransportConfig::FileInput(config) => Box::new(FileInputEndpoint::new(config)), - #[cfg(feature = "with-kafka")] - TransportConfig::KafkaInput(config) => Box::new(KafkaFtInputEndpoint::new(config)?), - #[cfg(not(feature = "with-kafka"))] - TransportConfig::KafkaInput(_) => return Ok(None), - #[cfg(feature = "with-nats")] - TransportConfig::NatsInput(config) => Box::new(NatsInputEndpoint::new(config)?), - #[cfg(not(feature = "with-nats"))] - TransportConfig::NatsInput(_) => return Ok(None), - #[cfg(feature = "with-pubsub")] - TransportConfig::PubSubInput(config) => Box::new(PubSubInputEndpoint::new(config.clone())?), - #[cfg(not(feature = "with-pubsub"))] - TransportConfig::PubSubInput(_) => return Ok(None), - TransportConfig::UrlInput(config) => Box::new(UrlInputEndpoint::new(config)), - TransportConfig::S3Input(config) => Box::new(S3InputEndpoint::new(config)?), - TransportConfig::Datagen(config) => Box::new(GeneratorEndpoint::new(config.clone())), - #[cfg(feature = "with-nexmark")] - TransportConfig::Nexmark(config) => Box::new(NexmarkEndpoint::new(config.clone())), - #[cfg(not(feature = "with-nexmark"))] - TransportConfig::Nexmark(_) => return Ok(None), - TransportConfig::HttpInput(config) => Box::new(HttpInputEndpoint::new(config)), - TransportConfig::AdHocInput(config) => Box::new(AdHocInputEndpoint::new(config)), - TransportConfig::ClockInput(config) => Box::new(ClockEndpoint::new(config)?), - TransportConfig::EmptyInput => Box::new(EmptyInputEndpoint), - TransportConfig::FileOutput(_) - | TransportConfig::KafkaOutput(_) - | TransportConfig::DeltaTableInput(_) - | TransportConfig::DeltaTableOutput(_) - | TransportConfig::DynamoDBOutput(_) - | TransportConfig::PostgresInput(_) - | TransportConfig::PostgresCdcInput(_) - | TransportConfig::PostgresOutput(_) - | TransportConfig::HttpOutput(_) - | TransportConfig::RedisOutput(_) - | TransportConfig::IcebergInput(_) - | TransportConfig::NullOutput => return Ok(None), - }; - Ok(Some(endpoint)) + builtin_input_transport_registry().create_endpoint(&config) } /// Creates an output transport endpoint instance using an output transport @@ -150,21 +414,101 @@ pub fn output_transport_config_to_endpoint( secrets_dir: &Path, ) -> AnyResult>> { let config = resolve_secret_references_via_json(secrets_dir, config)?; - match config { - TransportConfig::FileOutput(config) => Ok(Some(Box::new(FileOutputEndpoint::new(config)?))), - #[cfg(feature = "with-kafka")] - TransportConfig::KafkaOutput(config) => match fault_tolerant { - false => Ok(Some(Box::new(KafkaOutputEndpoint::new( - config, - endpoint_name, - )?))), - true => Ok(Some(Box::new(KafkaFtOutputEndpoint::new(config)?))), - }, - #[cfg(feature = "with-redis")] - TransportConfig::RedisOutput(config) => { - Ok(Some(Box::new(RedisOutputEndpoint::new(config)?))) - } - TransportConfig::NullOutput => Ok(Some(Box::new(NullOutputEndpoint))), - _ => Ok(None), + builtin_output_transport_registry().create_endpoint(&config, endpoint_name, fault_tolerant) +} + +#[cfg(test)] +mod tests { + use super::*; + use feldera_types::config::FtModel; + + #[test] + fn builtin_input_registry_creates_empty_input_endpoint() { + let secrets_dir = tempfile::tempdir().unwrap(); + + let endpoint = input_transport_config_to_endpoint( + &TransportConfig::EmptyInput, + "empty", + secrets_dir.path(), + ) + .unwrap() + .unwrap(); + + assert_eq!(endpoint.fault_tolerance(), Some(FtModel::ExactlyOnce)); + } + + #[test] + fn builtin_output_registry_creates_null_output_endpoint() { + let secrets_dir = tempfile::tempdir().unwrap(); + + let endpoint = output_transport_config_to_endpoint( + &TransportConfig::NullOutput, + "null", + true, + secrets_dir.path(), + ) + .unwrap() + .unwrap(); + + assert!(endpoint.is_fault_tolerant()); + assert_eq!(endpoint.max_buffer_size_bytes(), usize::MAX); + } + + #[test] + fn wrong_direction_transport_configs_still_return_none() { + let secrets_dir = tempfile::tempdir().unwrap(); + + assert!( + input_transport_config_to_endpoint( + &TransportConfig::NullOutput, + "null", + secrets_dir.path() + ) + .unwrap() + .is_none() + ); + assert!( + output_transport_config_to_endpoint( + &TransportConfig::EmptyInput, + "empty", + false, + secrets_dir.path() + ) + .unwrap() + .is_none() + ); + } + + #[test] + fn explicit_transport_registries_dispatch_by_transport_name() { + let mut input_registry = InputTransportRegistry::new(); + assert!( + input_registry + .create_endpoint(&TransportConfig::EmptyInput) + .unwrap() + .is_none() + ); + input_registry.register("empty_input", Box::new(EmptyInputFactory)); + assert!( + input_registry + .create_endpoint(&TransportConfig::EmptyInput) + .unwrap() + .is_some() + ); + + let mut output_registry = OutputTransportRegistry::new(); + assert!( + output_registry + .create_endpoint(&TransportConfig::NullOutput, "null", true) + .unwrap() + .is_none() + ); + output_registry.register("null_output", Box::new(NullOutputFactory)); + assert!( + output_registry + .create_endpoint(&TransportConfig::NullOutput, "null", true) + .unwrap() + .is_some() + ); } } From fec8f52dfeba478ae13b30849f663b106e2073cd Mon Sep 17 00:00:00 2001 From: Dnreikronos Date: Mon, 22 Jun 2026 17:57:25 -0300 Subject: [PATCH 2/4] Tighten transport registry scope --- crates/adapters/src/integrated.rs | 480 ++++------------------------- crates/adapters/src/transport.rs | 65 ++-- crates/feldera-types/src/config.rs | 73 +++-- 3 files changed, 161 insertions(+), 457 deletions(-) diff --git a/crates/adapters/src/integrated.rs b/crates/adapters/src/integrated.rs index baea053d1d3..a36df0f8d3d 100644 --- a/crates/adapters/src/integrated.rs +++ b/crates/adapters/src/integrated.rs @@ -4,10 +4,7 @@ use crate::{ControllerError, Encoder, InputConsumer, OutputEndpoint}; use datafusion::execution::runtime_env::RuntimeEnv; use feldera_types::config::{ConnectorConfig, PipelineConfig, TransportConfig}; use feldera_types::program_schema::Relation; -use std::{ - collections::BTreeMap, - sync::{Arc, Weak}, -}; +use std::sync::{Arc, Weak}; #[cfg(feature = "with-deltalake")] pub mod delta_table; @@ -43,354 +40,6 @@ where } } -/// Factory for creating integrated output endpoints from connector configuration. -pub trait IntegratedOutputEndpointFactory: Send + Sync { - #[allow(clippy::too_many_arguments)] - fn create( - &self, - endpoint_id: EndpointId, - endpoint_name: &str, - connector_config: &ConnectorConfig, - key_schema: &Option, - schema: &Relation, - controller: Weak, - continue_previous_state: bool, - is_index: bool, - ) -> Result>, ControllerError>; -} - -/// Registry of integrated output endpoint factories keyed by transport name. -#[derive(Default)] -pub struct IntegratedOutputEndpointRegistry { - registered: BTreeMap<&'static str, Arc>, -} - -impl IntegratedOutputEndpointRegistry { - pub fn new() -> Self { - Self { - registered: BTreeMap::new(), - } - } - - pub fn register( - &mut self, - name: &'static str, - factory: Box, - ) { - self.registered.insert(name, Arc::from(factory)); - } - - pub fn get(&self, name: &str) -> Option> { - self.registered.get(name).cloned() - } - - #[allow(clippy::too_many_arguments)] - pub fn create_endpoint( - &self, - endpoint_id: EndpointId, - endpoint_name: &str, - connector_config: &ConnectorConfig, - key_schema: &Option, - schema: &Relation, - controller: Weak, - continue_previous_state: bool, - is_index: bool, - ) -> Result>, ControllerError> { - let Some(factory) = self.get(&connector_config.transport.name()) else { - return Ok(None); - }; - factory.create( - endpoint_id, - endpoint_name, - connector_config, - key_schema, - schema, - controller, - continue_previous_state, - is_index, - ) - } -} - -/// Factory for creating integrated input endpoints from connector configuration. -pub trait IntegratedInputEndpointFactory: Send + Sync { - fn create( - &self, - endpoint_name: &str, - config: &ConnectorConfig, - pipeline_config: &PipelineConfig, - runtime_env: Arc, - consumer: Box, - ) -> Result>, ControllerError>; -} - -/// Registry of integrated input endpoint factories keyed by transport name. -#[derive(Default)] -pub struct IntegratedInputEndpointRegistry { - registered: BTreeMap<&'static str, Arc>, -} - -impl IntegratedInputEndpointRegistry { - pub fn new() -> Self { - Self { - registered: BTreeMap::new(), - } - } - - pub fn register( - &mut self, - name: &'static str, - factory: Box, - ) { - self.registered.insert(name, Arc::from(factory)); - } - - pub fn get(&self, name: &str) -> Option> { - self.registered.get(name).cloned() - } - - pub fn create_endpoint( - &self, - endpoint_name: &str, - config: &ConnectorConfig, - pipeline_config: &PipelineConfig, - runtime_env: Arc, - consumer: Box, - ) -> Result>, ControllerError> { - let Some(factory) = self.get(&config.transport.name()) else { - return Ok(None); - }; - factory.create( - endpoint_name, - config, - pipeline_config, - runtime_env, - consumer, - ) - } -} - -pub fn builtin_integrated_output_endpoint_registry() -> IntegratedOutputEndpointRegistry { - let mut registry = IntegratedOutputEndpointRegistry::new(); - #[cfg(feature = "with-deltalake")] - registry.register("delta_table_output", Box::new(DeltaTableOutputFactory)); - registry.register("postgres_output", Box::new(PostgresOutputFactory)); - #[cfg(feature = "with-dynamodb")] - registry.register("dynamodb_output", Box::new(DynamoDBOutputFactory)); - registry -} - -pub fn builtin_integrated_input_endpoint_registry() -> IntegratedInputEndpointRegistry { - let mut registry = IntegratedInputEndpointRegistry::new(); - #[cfg(feature = "with-deltalake")] - registry.register("delta_table_input", Box::new(DeltaTableInputFactory)); - #[cfg(feature = "with-iceberg")] - registry.register("iceberg_input", Box::new(IcebergInputFactory)); - registry.register("postgres_input", Box::new(PostgresInputFactory)); - #[cfg(feature = "with-postgres-cdc")] - registry.register("postgres_cdc_input", Box::new(PostgresCdcInputFactory)); - registry -} - -#[cfg(feature = "with-deltalake")] -struct DeltaTableOutputFactory; - -#[cfg(feature = "with-deltalake")] -impl IntegratedOutputEndpointFactory for DeltaTableOutputFactory { - #[allow(clippy::too_many_arguments)] - fn create( - &self, - endpoint_id: EndpointId, - endpoint_name: &str, - connector_config: &ConnectorConfig, - key_schema: &Option, - schema: &Relation, - controller: Weak, - continue_previous_state: bool, - is_index: bool, - ) -> Result>, ControllerError> { - match &connector_config.transport { - TransportConfig::DeltaTableOutput(config) => { - Ok(Some(Box::new(delta_table::DeltaTableWriter::new( - endpoint_id, - endpoint_name, - config, - key_schema, - schema, - controller, - continue_previous_state, - is_index, - )?))) - } - _ => Ok(None), - } - } -} - -struct PostgresOutputFactory; - -impl IntegratedOutputEndpointFactory for PostgresOutputFactory { - #[allow(clippy::too_many_arguments)] - fn create( - &self, - endpoint_id: EndpointId, - endpoint_name: &str, - connector_config: &ConnectorConfig, - key_schema: &Option, - schema: &Relation, - controller: Weak, - _continue_previous_state: bool, - is_index: bool, - ) -> Result>, ControllerError> { - match &connector_config.transport { - TransportConfig::PostgresOutput(config) => { - Ok(Some(Box::new(PostgresOutputEndpoint::new( - endpoint_id, - endpoint_name, - config, - key_schema, - schema, - controller, - is_index, - )?))) - } - _ => Ok(None), - } - } -} - -#[cfg(feature = "with-dynamodb")] -struct DynamoDBOutputFactory; - -#[cfg(feature = "with-dynamodb")] -impl IntegratedOutputEndpointFactory for DynamoDBOutputFactory { - #[allow(clippy::too_many_arguments)] - fn create( - &self, - endpoint_id: EndpointId, - endpoint_name: &str, - connector_config: &ConnectorConfig, - key_schema: &Option, - schema: &Relation, - controller: Weak, - _continue_previous_state: bool, - is_index: bool, - ) -> Result>, ControllerError> { - match &connector_config.transport { - TransportConfig::DynamoDBOutput(config) => { - Ok(Some(Box::new(DynamoDBOutputEndpoint::new( - endpoint_id, - endpoint_name, - config, - key_schema, - schema, - controller, - is_index, - )?))) - } - _ => Ok(None), - } - } -} - -#[cfg(feature = "with-deltalake")] -struct DeltaTableInputFactory; - -#[cfg(feature = "with-deltalake")] -impl IntegratedInputEndpointFactory for DeltaTableInputFactory { - fn create( - &self, - endpoint_name: &str, - config: &ConnectorConfig, - pipeline_config: &PipelineConfig, - runtime_env: Arc, - consumer: Box, - ) -> Result>, ControllerError> { - match &config.transport { - TransportConfig::DeltaTableInput(config) => { - Ok(Some(Box::new(delta_table::DeltaTableInputEndpoint::new( - endpoint_name, - config, - pipeline_config, - runtime_env, - consumer, - )))) - } - _ => Ok(None), - } - } -} - -#[cfg(feature = "with-iceberg")] -struct IcebergInputFactory; - -#[cfg(feature = "with-iceberg")] -impl IntegratedInputEndpointFactory for IcebergInputFactory { - fn create( - &self, - endpoint_name: &str, - config: &ConnectorConfig, - pipeline_config: &PipelineConfig, - runtime_env: Arc, - consumer: Box, - ) -> Result>, ControllerError> { - match &config.transport { - TransportConfig::IcebergInput(config) => { - Ok(Some(Box::new(feldera_iceberg::IcebergInputEndpoint::new( - endpoint_name, - config, - pipeline_config, - runtime_env, - consumer, - )))) - } - _ => Ok(None), - } - } -} - -struct PostgresInputFactory; - -impl IntegratedInputEndpointFactory for PostgresInputFactory { - fn create( - &self, - endpoint_name: &str, - config: &ConnectorConfig, - _pipeline_config: &PipelineConfig, - _runtime_env: Arc, - consumer: Box, - ) -> Result>, ControllerError> { - match &config.transport { - TransportConfig::PostgresInput(config) => Ok(Some(Box::new( - PostgresInputEndpoint::new(endpoint_name, config, consumer), - ))), - _ => Ok(None), - } - } -} - -#[cfg(feature = "with-postgres-cdc")] -struct PostgresCdcInputFactory; - -#[cfg(feature = "with-postgres-cdc")] -impl IntegratedInputEndpointFactory for PostgresCdcInputFactory { - fn create( - &self, - endpoint_name: &str, - config: &ConnectorConfig, - _pipeline_config: &PipelineConfig, - _runtime_env: Arc, - consumer: Box, - ) -> Result>, ControllerError> { - match &config.transport { - TransportConfig::PostgresCdcInput(config) => Ok(Some(Box::new( - PostgresCdcInputEndpoint::new(endpoint_name, config, consumer), - ))), - _ => Ok(None), - } - } -} - /// Create an instance of an integrated output endpoint given its config /// and output relation schema. #[allow(unused, clippy::too_many_arguments)] @@ -404,49 +53,44 @@ pub fn create_integrated_output_endpoint( continue_previous_state: bool, is_index: bool, ) -> Result, ControllerError> { - let registry = builtin_integrated_output_endpoint_registry(); - create_integrated_output_endpoint_with_registry( - ®istry, - endpoint_id, - endpoint_name, - connector_config, - key_schema, - schema, - controller, - continue_previous_state, - is_index, - ) -} - -#[allow(unused, clippy::too_many_arguments)] -pub fn create_integrated_output_endpoint_with_registry( - registry: &IntegratedOutputEndpointRegistry, - endpoint_id: EndpointId, - endpoint_name: &str, - connector_config: &ConnectorConfig, - key_schema: &Option, - schema: &Relation, - controller: Weak, - continue_previous_state: bool, - is_index: bool, -) -> Result, ControllerError> { - let ep = registry - .create_endpoint( + let ep: Box = match &connector_config.transport { + #[cfg(feature = "with-deltalake")] + TransportConfig::DeltaTableOutput(config) => Box::new(delta_table::DeltaTableWriter::new( endpoint_id, endpoint_name, - connector_config, + config, key_schema, schema, controller, continue_previous_state, is_index, - )? - .ok_or_else(|| { - ControllerError::unknown_output_transport( + )?), + TransportConfig::PostgresOutput(config) => Box::new(PostgresOutputEndpoint::new( + endpoint_id, + endpoint_name, + config, + key_schema, + schema, + controller, + is_index, + )?), + #[cfg(feature = "with-dynamodb")] + TransportConfig::DynamoDBOutput(config) => Box::new(DynamoDBOutputEndpoint::new( + endpoint_id, + endpoint_name, + config, + key_schema, + schema, + controller, + is_index, + )?), + transport => { + return Err(ControllerError::unknown_output_transport( endpoint_name, - &connector_config.transport.name(), - ) - })?; + &transport.name(), + )); + } + }; if connector_config.format.is_some() { return Err(ControllerError::invalid_parser_configuration( @@ -469,37 +113,43 @@ pub fn create_integrated_input_endpoint( runtime_env: Arc, consumer: Box, ) -> Result, ControllerError> { - let registry = builtin_integrated_input_endpoint_registry(); - create_integrated_input_endpoint_with_registry( - ®istry, - endpoint_name, - config, - pipeline_config, - runtime_env, - consumer, - ) -} - -#[allow(unused_variables)] -pub fn create_integrated_input_endpoint_with_registry( - registry: &IntegratedInputEndpointRegistry, - endpoint_name: &str, - config: &ConnectorConfig, - pipeline_config: &PipelineConfig, - runtime_env: Arc, - consumer: Box, -) -> Result, ControllerError> { - let ep = registry - .create_endpoint( + let ep: Box = match &config.transport { + #[cfg(feature = "with-deltalake")] + TransportConfig::DeltaTableInput(config) => { + Box::new(delta_table::DeltaTableInputEndpoint::new( + endpoint_name, + config, + pipeline_config, + runtime_env, + consumer, + )) + } + #[cfg(feature = "with-iceberg")] + TransportConfig::IcebergInput(config) => { + Box::new(feldera_iceberg::IcebergInputEndpoint::new( + endpoint_name, + config, + pipeline_config, + runtime_env, + consumer, + )) + } + TransportConfig::PostgresInput(config) => { + Box::new(PostgresInputEndpoint::new(endpoint_name, config, consumer)) + } + #[cfg(feature = "with-postgres-cdc")] + TransportConfig::PostgresCdcInput(config) => Box::new(PostgresCdcInputEndpoint::new( endpoint_name, config, - pipeline_config, - runtime_env, consumer, - )? - .ok_or_else(|| { - ControllerError::unknown_input_transport(endpoint_name, &config.transport.name()) - })?; + )), + transport => { + return Err(ControllerError::unknown_input_transport( + endpoint_name, + &transport.name(), + )); + } + }; if config.format.is_some() { return Err(ControllerError::invalid_parser_configuration( diff --git a/crates/adapters/src/transport.rs b/crates/adapters/src/transport.rs index d5ad42d7b21..61048cf3dee 100644 --- a/crates/adapters/src/transport.rs +++ b/crates/adapters/src/transport.rs @@ -82,33 +82,33 @@ pub use feldera_adapterlib::transport::*; pub fn builtin_input_transport_registry() -> InputTransportRegistry { let mut registry = InputTransportRegistry::new(); - registry.register("file_input", Box::new(FileInputFactory)); + registry.register(TransportConfig::FILE_INPUT, Box::new(FileInputFactory)); #[cfg(feature = "with-kafka")] - registry.register("kafka_input", Box::new(KafkaInputFactory)); + registry.register(TransportConfig::KAFKA_INPUT, Box::new(KafkaInputFactory)); #[cfg(feature = "with-nats")] - registry.register("nats_input", Box::new(NatsInputFactory)); + registry.register(TransportConfig::NATS_INPUT, Box::new(NatsInputFactory)); #[cfg(feature = "with-pubsub")] - registry.register("pub_sub_input", Box::new(PubSubInputFactory)); - registry.register("url_input", Box::new(UrlInputFactory)); - registry.register("s3_input", Box::new(S3InputFactory)); - registry.register("datagen", Box::new(DatagenInputFactory)); + registry.register(TransportConfig::PUB_SUB_INPUT, Box::new(PubSubInputFactory)); + registry.register(TransportConfig::URL_INPUT, Box::new(UrlInputFactory)); + registry.register(TransportConfig::S3_INPUT, Box::new(S3InputFactory)); + registry.register(TransportConfig::DATAGEN, Box::new(DatagenInputFactory)); #[cfg(feature = "with-nexmark")] - registry.register("nexmark", Box::new(NexmarkInputFactory)); - registry.register("http_input", Box::new(HttpInputFactory)); - registry.register("adhoc_input", Box::new(AdHocInputFactory)); - registry.register("clock", Box::new(ClockInputFactory)); - registry.register("empty_input", Box::new(EmptyInputFactory)); + registry.register(TransportConfig::NEXMARK, Box::new(NexmarkInputFactory)); + registry.register(TransportConfig::HTTP_INPUT, Box::new(HttpInputFactory)); + registry.register(TransportConfig::ADHOC_INPUT, Box::new(AdHocInputFactory)); + registry.register(TransportConfig::CLOCK, Box::new(ClockInputFactory)); + registry.register(TransportConfig::EMPTY_INPUT, Box::new(EmptyInputFactory)); registry } pub fn builtin_output_transport_registry() -> OutputTransportRegistry { let mut registry = OutputTransportRegistry::new(); - registry.register("file_output", Box::new(FileOutputFactory)); + registry.register(TransportConfig::FILE_OUTPUT, Box::new(FileOutputFactory)); #[cfg(feature = "with-kafka")] - registry.register("kafka_output", Box::new(KafkaOutputFactory)); + registry.register(TransportConfig::KAFKA_OUTPUT, Box::new(KafkaOutputFactory)); #[cfg(feature = "with-redis")] - registry.register("redis_output", Box::new(RedisOutputFactory)); - registry.register("null_output", Box::new(NullOutputFactory)); + registry.register(TransportConfig::REDIS_OUTPUT, Box::new(RedisOutputFactory)); + registry.register(TransportConfig::NULL_OUTPUT, Box::new(NullOutputFactory)); registry } @@ -488,7 +488,7 @@ mod tests { .unwrap() .is_none() ); - input_registry.register("empty_input", Box::new(EmptyInputFactory)); + input_registry.register(TransportConfig::EMPTY_INPUT, Box::new(EmptyInputFactory)); assert!( input_registry .create_endpoint(&TransportConfig::EmptyInput) @@ -503,7 +503,7 @@ mod tests { .unwrap() .is_none() ); - output_registry.register("null_output", Box::new(NullOutputFactory)); + output_registry.register(TransportConfig::NULL_OUTPUT, Box::new(NullOutputFactory)); assert!( output_registry .create_endpoint(&TransportConfig::NullOutput, "null", true) @@ -511,4 +511,33 @@ mod tests { .is_some() ); } + + #[test] + fn builtin_transport_registries_include_compiled_transport_names() { + let input_registry = builtin_input_transport_registry(); + assert!(input_registry.get(TransportConfig::FILE_INPUT).is_some()); + #[cfg(feature = "with-kafka")] + assert!(input_registry.get(TransportConfig::KAFKA_INPUT).is_some()); + #[cfg(feature = "with-nats")] + assert!(input_registry.get(TransportConfig::NATS_INPUT).is_some()); + #[cfg(feature = "with-pubsub")] + assert!(input_registry.get(TransportConfig::PUB_SUB_INPUT).is_some()); + assert!(input_registry.get(TransportConfig::URL_INPUT).is_some()); + assert!(input_registry.get(TransportConfig::S3_INPUT).is_some()); + assert!(input_registry.get(TransportConfig::DATAGEN).is_some()); + #[cfg(feature = "with-nexmark")] + assert!(input_registry.get(TransportConfig::NEXMARK).is_some()); + assert!(input_registry.get(TransportConfig::HTTP_INPUT).is_some()); + assert!(input_registry.get(TransportConfig::ADHOC_INPUT).is_some()); + assert!(input_registry.get(TransportConfig::CLOCK).is_some()); + assert!(input_registry.get(TransportConfig::EMPTY_INPUT).is_some()); + + let output_registry = builtin_output_transport_registry(); + assert!(output_registry.get(TransportConfig::FILE_OUTPUT).is_some()); + #[cfg(feature = "with-kafka")] + assert!(output_registry.get(TransportConfig::KAFKA_OUTPUT).is_some()); + #[cfg(feature = "with-redis")] + assert!(output_registry.get(TransportConfig::REDIS_OUTPUT).is_some()); + assert!(output_registry.get(TransportConfig::NULL_OUTPUT).is_some()); + } } diff --git a/crates/feldera-types/src/config.rs b/crates/feldera-types/src/config.rs index ad34a38e74d..e5ff58c3687 100644 --- a/crates/feldera-types/src/config.rs +++ b/crates/feldera-types/src/config.rs @@ -1861,32 +1861,57 @@ pub enum TransportConfig { } impl TransportConfig { + pub const FILE_INPUT: &'static str = "file_input"; + pub const FILE_OUTPUT: &'static str = "file_output"; + pub const NATS_INPUT: &'static str = "nats_input"; + pub const KAFKA_INPUT: &'static str = "kafka_input"; + pub const KAFKA_OUTPUT: &'static str = "kafka_output"; + pub const PUB_SUB_INPUT: &'static str = "pub_sub_input"; + pub const URL_INPUT: &'static str = "url_input"; + pub const S3_INPUT: &'static str = "s3_input"; + pub const DELTA_TABLE_INPUT: &'static str = "delta_table_input"; + pub const DELTA_TABLE_OUTPUT: &'static str = "delta_table_output"; + pub const DYNAMODB_OUTPUT: &'static str = "dynamodb_output"; + pub const REDIS_OUTPUT: &'static str = "redis_output"; + pub const ICEBERG_INPUT: &'static str = "iceberg_input"; + pub const POSTGRES_INPUT: &'static str = "postgres_input"; + pub const POSTGRES_CDC_INPUT: &'static str = "postgres_cdc_input"; + pub const POSTGRES_OUTPUT: &'static str = "postgres_output"; + pub const DATAGEN: &'static str = "datagen"; + pub const NEXMARK: &'static str = "nexmark"; + pub const HTTP_INPUT: &'static str = "http_input"; + pub const HTTP_OUTPUT: &'static str = "http_output"; + pub const ADHOC_INPUT: &'static str = "adhoc_input"; + pub const CLOCK: &'static str = "clock"; + pub const NULL_OUTPUT: &'static str = "null_output"; + pub const EMPTY_INPUT: &'static str = "empty_input"; + pub fn name(&self) -> String { match self { - TransportConfig::FileInput(_) => "file_input".to_string(), - TransportConfig::FileOutput(_) => "file_output".to_string(), - TransportConfig::NatsInput(_) => "nats_input".to_string(), - TransportConfig::KafkaInput(_) => "kafka_input".to_string(), - TransportConfig::KafkaOutput(_) => "kafka_output".to_string(), - TransportConfig::PubSubInput(_) => "pub_sub_input".to_string(), - TransportConfig::UrlInput(_) => "url_input".to_string(), - TransportConfig::S3Input(_) => "s3_input".to_string(), - TransportConfig::DeltaTableInput(_) => "delta_table_input".to_string(), - TransportConfig::DeltaTableOutput(_) => "delta_table_output".to_string(), - TransportConfig::DynamoDBOutput(_) => "dynamodb_output".to_string(), - TransportConfig::IcebergInput(_) => "iceberg_input".to_string(), - TransportConfig::PostgresInput(_) => "postgres_input".to_string(), - TransportConfig::PostgresCdcInput(_) => "postgres_cdc_input".to_string(), - TransportConfig::PostgresOutput(_) => "postgres_output".to_string(), - TransportConfig::Datagen(_) => "datagen".to_string(), - TransportConfig::Nexmark(_) => "nexmark".to_string(), - TransportConfig::HttpInput(_) => "http_input".to_string(), - TransportConfig::HttpOutput(_) => "http_output".to_string(), - TransportConfig::AdHocInput(_) => "adhoc_input".to_string(), - TransportConfig::RedisOutput(_) => "redis_output".to_string(), - TransportConfig::ClockInput(_) => "clock".to_string(), - TransportConfig::NullOutput => "null_output".to_string(), - TransportConfig::EmptyInput => "empty_input".to_string(), + TransportConfig::FileInput(_) => Self::FILE_INPUT.to_string(), + TransportConfig::FileOutput(_) => Self::FILE_OUTPUT.to_string(), + TransportConfig::NatsInput(_) => Self::NATS_INPUT.to_string(), + TransportConfig::KafkaInput(_) => Self::KAFKA_INPUT.to_string(), + TransportConfig::KafkaOutput(_) => Self::KAFKA_OUTPUT.to_string(), + TransportConfig::PubSubInput(_) => Self::PUB_SUB_INPUT.to_string(), + TransportConfig::UrlInput(_) => Self::URL_INPUT.to_string(), + TransportConfig::S3Input(_) => Self::S3_INPUT.to_string(), + TransportConfig::DeltaTableInput(_) => Self::DELTA_TABLE_INPUT.to_string(), + TransportConfig::DeltaTableOutput(_) => Self::DELTA_TABLE_OUTPUT.to_string(), + TransportConfig::DynamoDBOutput(_) => Self::DYNAMODB_OUTPUT.to_string(), + TransportConfig::IcebergInput(_) => Self::ICEBERG_INPUT.to_string(), + TransportConfig::PostgresInput(_) => Self::POSTGRES_INPUT.to_string(), + TransportConfig::PostgresCdcInput(_) => Self::POSTGRES_CDC_INPUT.to_string(), + TransportConfig::PostgresOutput(_) => Self::POSTGRES_OUTPUT.to_string(), + TransportConfig::Datagen(_) => Self::DATAGEN.to_string(), + TransportConfig::Nexmark(_) => Self::NEXMARK.to_string(), + TransportConfig::HttpInput(_) => Self::HTTP_INPUT.to_string(), + TransportConfig::HttpOutput(_) => Self::HTTP_OUTPUT.to_string(), + TransportConfig::AdHocInput(_) => Self::ADHOC_INPUT.to_string(), + TransportConfig::RedisOutput(_) => Self::REDIS_OUTPUT.to_string(), + TransportConfig::ClockInput(_) => Self::CLOCK.to_string(), + TransportConfig::NullOutput => Self::NULL_OUTPUT.to_string(), + TransportConfig::EmptyInput => Self::EMPTY_INPUT.to_string(), } } From 2e5be663dd79229c13cdc869625b9eb53b7dd6b7 Mon Sep 17 00:00:00 2001 From: Dnreikronos Date: Wed, 24 Jun 2026 11:48:44 -0300 Subject: [PATCH 3/4] Allow dynamic transport registry keys --- crates/adapterlib/src/transport.rs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/crates/adapterlib/src/transport.rs b/crates/adapterlib/src/transport.rs index 7225408b055..56416bf8c98 100644 --- a/crates/adapterlib/src/transport.rs +++ b/crates/adapterlib/src/transport.rs @@ -86,7 +86,7 @@ pub trait InputTransportEndpointFactory: Send + Sync { /// Registry of input transport endpoint factories keyed by transport name. #[derive(Default)] pub struct InputTransportRegistry { - registered: BTreeMap<&'static str, Arc>, + registered: BTreeMap>, } impl InputTransportRegistry { @@ -98,10 +98,10 @@ impl InputTransportRegistry { pub fn register( &mut self, - name: &'static str, + name: impl Into, factory: Box, ) { - self.registered.insert(name, Arc::from(factory)); + self.registered.insert(name.into(), Arc::from(factory)); } pub fn get(&self, name: &str) -> Option> { @@ -112,7 +112,7 @@ impl InputTransportRegistry { &self, config: &TransportConfig, ) -> AnyResult>> { - let Some(factory) = self.get(&config.name()) else { + let Some(factory) = self.get(config.name().as_str()) else { return Ok(None); }; factory.create(config) @@ -1151,7 +1151,7 @@ pub trait OutputTransportEndpointFactory: Send + Sync { /// Registry of output transport endpoint factories keyed by transport name. #[derive(Default)] pub struct OutputTransportRegistry { - registered: BTreeMap<&'static str, Arc>, + registered: BTreeMap>, } impl OutputTransportRegistry { @@ -1163,10 +1163,10 @@ impl OutputTransportRegistry { pub fn register( &mut self, - name: &'static str, + name: impl Into, factory: Box, ) { - self.registered.insert(name, Arc::from(factory)); + self.registered.insert(name.into(), Arc::from(factory)); } pub fn get(&self, name: &str) -> Option> { @@ -1179,7 +1179,7 @@ impl OutputTransportRegistry { endpoint_name: &str, fault_tolerant: bool, ) -> AnyResult>> { - let Some(factory) = self.get(&config.name()) else { + let Some(factory) = self.get(config.name().as_str()) else { return Ok(None); }; factory.create(config, endpoint_name, fault_tolerant) From 3f43c20fcd64918a77520296c3dad58e5af4ebb0 Mon Sep 17 00:00:00 2001 From: Dnreikronos Date: Fri, 3 Jul 2026 09:15:01 -0300 Subject: [PATCH 4/4] Report missing transport factories --- crates/adapterlib/src/transport.rs | 13 +- crates/adapters/src/controller.rs | 42 +++- crates/adapters/src/transport.rs | 356 ++++++++++++++++++++--------- 3 files changed, 281 insertions(+), 130 deletions(-) diff --git a/crates/adapterlib/src/transport.rs b/crates/adapterlib/src/transport.rs index 56416bf8c98..b0459566156 100644 --- a/crates/adapterlib/src/transport.rs +++ b/crates/adapterlib/src/transport.rs @@ -77,10 +77,7 @@ pub trait TransportInputEndpoint: InputEndpoint { /// Factory for creating input transport endpoints from transport configuration. pub trait InputTransportEndpointFactory: Send + Sync { - fn create( - &self, - config: &TransportConfig, - ) -> AnyResult>>; + fn create(&self, config: &TransportConfig) -> AnyResult>; } /// Registry of input transport endpoint factories keyed by transport name. @@ -115,7 +112,7 @@ impl InputTransportRegistry { let Some(factory) = self.get(config.name().as_str()) else { return Ok(None); }; - factory.create(config) + factory.create(config).map(Some) } } @@ -1145,7 +1142,7 @@ pub trait OutputTransportEndpointFactory: Send + Sync { config: &TransportConfig, endpoint_name: &str, fault_tolerant: bool, - ) -> AnyResult>>; + ) -> AnyResult>; } /// Registry of output transport endpoint factories keyed by transport name. @@ -1182,7 +1179,9 @@ impl OutputTransportRegistry { let Some(factory) = self.get(config.name().as_str()) else { return Ok(None); }; - factory.create(config, endpoint_name, fault_tolerant) + factory + .create(config, endpoint_name, fault_tolerant) + .map(Some) } } diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index 8864a11540f..866060ca349 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -33,8 +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::{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, @@ -6274,16 +6274,25 @@ impl ControllerInner { &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_config.name()); + .get(&transport_name); let endpoint = match factory { - Some(factory) => factory - .create(&transport_config) - .map_err(|e| ControllerError::input_transport_error(endpoint_name, true, e))?, + 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, }; @@ -6605,20 +6614,29 @@ impl ControllerInner { &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_config.name()); + .get(&transport_name); let endpoint = match factory { - Some(factory) => factory - .create( - &transport_config, + 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, - self.fault_tolerance == Some(FtModel::ExactlyOnce), - ) - .map_err(|e| ControllerError::output_transport_error(endpoint_name, true, e))?, + &transport_name, + )); + } None => None, }; diff --git a/crates/adapters/src/transport.rs b/crates/adapters/src/transport.rs index 61048cf3dee..d106b92d23d 100644 --- a/crates/adapters/src/transport.rs +++ b/crates/adapters/src/transport.rs @@ -23,7 +23,7 @@ use std::path::Path; use adhoc::AdHocInputEndpoint; -use anyhow::Result as AnyResult; +use anyhow::{Result as AnyResult, anyhow}; use clock::ClockEndpoint; use feldera_types::secret_resolver::resolve_secret_references_via_json; use http::HttpInputEndpoint; @@ -112,18 +112,61 @@ pub fn builtin_output_transport_registry() -> OutputTransportRegistry { registry } +pub fn input_transport_uses_registry(config: &TransportConfig) -> bool { + matches!( + config, + TransportConfig::FileInput(_) + | TransportConfig::KafkaInput(_) + | TransportConfig::NatsInput(_) + | TransportConfig::PubSubInput(_) + | TransportConfig::UrlInput(_) + | TransportConfig::S3Input(_) + | TransportConfig::Datagen(_) + | TransportConfig::Nexmark(_) + | TransportConfig::HttpInput(_) + | TransportConfig::AdHocInput(_) + | TransportConfig::ClockInput(_) + | TransportConfig::EmptyInput + ) +} + +pub fn output_transport_uses_registry(config: &TransportConfig) -> bool { + matches!( + config, + TransportConfig::FileOutput(_) + | TransportConfig::KafkaOutput(_) + | TransportConfig::RedisOutput(_) + | TransportConfig::NullOutput + ) +} + +fn unexpected_input_transport_config( + factory_name: &str, + config: &TransportConfig, +) -> AnyResult> { + Err(anyhow!( + "{factory_name} cannot create input endpoint for transport '{}'", + config.name() + )) +} + +fn unexpected_output_transport_config( + factory_name: &str, + config: &TransportConfig, +) -> AnyResult> { + Err(anyhow!( + "{factory_name} cannot create output endpoint for transport '{}'", + config.name() + )) +} + struct FileInputFactory; impl InputTransportEndpointFactory for FileInputFactory { - fn create( - &self, - config: &TransportConfig, - ) -> AnyResult>> { + fn create(&self, config: &TransportConfig) -> AnyResult> { match config { - TransportConfig::FileInput(config) => { - Ok(Some(Box::new(FileInputEndpoint::new(config)))) - } - _ => Ok(None), + TransportConfig::FileInput(config) => Ok(Box::new(FileInputEndpoint::new(config))), + _ => unexpected_input_transport_config("FileInputFactory", config), } } } @@ -133,15 +176,10 @@ struct KafkaInputFactory; #[cfg(feature = "with-kafka")] impl InputTransportEndpointFactory for KafkaInputFactory { - fn create( - &self, - config: &TransportConfig, - ) -> AnyResult>> { + fn create(&self, config: &TransportConfig) -> AnyResult> { match config { - TransportConfig::KafkaInput(config) => { - Ok(Some(Box::new(KafkaFtInputEndpoint::new(config)?))) - } - _ => Ok(None), + TransportConfig::KafkaInput(config) => Ok(Box::new(KafkaFtInputEndpoint::new(config)?)), + _ => unexpected_input_transport_config("KafkaInputFactory", config), } } } @@ -151,15 +189,10 @@ struct NatsInputFactory; #[cfg(feature = "with-nats")] impl InputTransportEndpointFactory for NatsInputFactory { - fn create( - &self, - config: &TransportConfig, - ) -> AnyResult>> { + fn create(&self, config: &TransportConfig) -> AnyResult> { match config { - TransportConfig::NatsInput(config) => { - Ok(Some(Box::new(NatsInputEndpoint::new(config)?))) - } - _ => Ok(None), + TransportConfig::NatsInput(config) => Ok(Box::new(NatsInputEndpoint::new(config)?)), + _ => unexpected_input_transport_config("NatsInputFactory", config), } } } @@ -169,15 +202,12 @@ struct PubSubInputFactory; #[cfg(feature = "with-pubsub")] impl InputTransportEndpointFactory for PubSubInputFactory { - fn create( - &self, - config: &TransportConfig, - ) -> AnyResult>> { + fn create(&self, config: &TransportConfig) -> AnyResult> { match config { TransportConfig::PubSubInput(config) => { - Ok(Some(Box::new(PubSubInputEndpoint::new(config.clone())?))) + Ok(Box::new(PubSubInputEndpoint::new(config.clone())?)) } - _ => Ok(None), + _ => unexpected_input_transport_config("PubSubInputFactory", config), } } } @@ -185,13 +215,10 @@ impl InputTransportEndpointFactory for PubSubInputFactory { struct UrlInputFactory; impl InputTransportEndpointFactory for UrlInputFactory { - fn create( - &self, - config: &TransportConfig, - ) -> AnyResult>> { + fn create(&self, config: &TransportConfig) -> AnyResult> { match config { - TransportConfig::UrlInput(config) => Ok(Some(Box::new(UrlInputEndpoint::new(config)))), - _ => Ok(None), + TransportConfig::UrlInput(config) => Ok(Box::new(UrlInputEndpoint::new(config))), + _ => unexpected_input_transport_config("UrlInputFactory", config), } } } @@ -199,13 +226,10 @@ impl InputTransportEndpointFactory for UrlInputFactory { struct S3InputFactory; impl InputTransportEndpointFactory for S3InputFactory { - fn create( - &self, - config: &TransportConfig, - ) -> AnyResult>> { + fn create(&self, config: &TransportConfig) -> AnyResult> { match config { - TransportConfig::S3Input(config) => Ok(Some(Box::new(S3InputEndpoint::new(config)?))), - _ => Ok(None), + TransportConfig::S3Input(config) => Ok(Box::new(S3InputEndpoint::new(config)?)), + _ => unexpected_input_transport_config("S3InputFactory", config), } } } @@ -213,15 +237,12 @@ impl InputTransportEndpointFactory for S3InputFactory { struct DatagenInputFactory; impl InputTransportEndpointFactory for DatagenInputFactory { - fn create( - &self, - config: &TransportConfig, - ) -> AnyResult>> { + fn create(&self, config: &TransportConfig) -> AnyResult> { match config { TransportConfig::Datagen(config) => { - Ok(Some(Box::new(GeneratorEndpoint::new(config.clone())))) + Ok(Box::new(GeneratorEndpoint::new(config.clone()))) } - _ => Ok(None), + _ => unexpected_input_transport_config("DatagenInputFactory", config), } } } @@ -231,15 +252,10 @@ struct NexmarkInputFactory; #[cfg(feature = "with-nexmark")] impl InputTransportEndpointFactory for NexmarkInputFactory { - fn create( - &self, - config: &TransportConfig, - ) -> AnyResult>> { + fn create(&self, config: &TransportConfig) -> AnyResult> { match config { - TransportConfig::Nexmark(config) => { - Ok(Some(Box::new(NexmarkEndpoint::new(config.clone())))) - } - _ => Ok(None), + TransportConfig::Nexmark(config) => Ok(Box::new(NexmarkEndpoint::new(config.clone()))), + _ => unexpected_input_transport_config("NexmarkInputFactory", config), } } } @@ -247,15 +263,10 @@ impl InputTransportEndpointFactory for NexmarkInputFactory { struct HttpInputFactory; impl InputTransportEndpointFactory for HttpInputFactory { - fn create( - &self, - config: &TransportConfig, - ) -> AnyResult>> { + fn create(&self, config: &TransportConfig) -> AnyResult> { match config { - TransportConfig::HttpInput(config) => { - Ok(Some(Box::new(HttpInputEndpoint::new(config)))) - } - _ => Ok(None), + TransportConfig::HttpInput(config) => Ok(Box::new(HttpInputEndpoint::new(config))), + _ => unexpected_input_transport_config("HttpInputFactory", config), } } } @@ -263,15 +274,10 @@ impl InputTransportEndpointFactory for HttpInputFactory { struct AdHocInputFactory; impl InputTransportEndpointFactory for AdHocInputFactory { - fn create( - &self, - config: &TransportConfig, - ) -> AnyResult>> { + fn create(&self, config: &TransportConfig) -> AnyResult> { match config { - TransportConfig::AdHocInput(config) => { - Ok(Some(Box::new(AdHocInputEndpoint::new(config)))) - } - _ => Ok(None), + TransportConfig::AdHocInput(config) => Ok(Box::new(AdHocInputEndpoint::new(config))), + _ => unexpected_input_transport_config("AdHocInputFactory", config), } } } @@ -279,13 +285,10 @@ impl InputTransportEndpointFactory for AdHocInputFactory { struct ClockInputFactory; impl InputTransportEndpointFactory for ClockInputFactory { - fn create( - &self, - config: &TransportConfig, - ) -> AnyResult>> { + fn create(&self, config: &TransportConfig) -> AnyResult> { match config { - TransportConfig::ClockInput(config) => Ok(Some(Box::new(ClockEndpoint::new(config)?))), - _ => Ok(None), + TransportConfig::ClockInput(config) => Ok(Box::new(ClockEndpoint::new(config)?)), + _ => unexpected_input_transport_config("ClockInputFactory", config), } } } @@ -293,13 +296,10 @@ impl InputTransportEndpointFactory for ClockInputFactory { struct EmptyInputFactory; impl InputTransportEndpointFactory for EmptyInputFactory { - fn create( - &self, - config: &TransportConfig, - ) -> AnyResult>> { + fn create(&self, config: &TransportConfig) -> AnyResult> { match config { - TransportConfig::EmptyInput => Ok(Some(Box::new(EmptyInputEndpoint))), - _ => Ok(None), + TransportConfig::EmptyInput => Ok(Box::new(EmptyInputEndpoint)), + _ => unexpected_input_transport_config("EmptyInputFactory", config), } } } @@ -312,12 +312,10 @@ impl OutputTransportEndpointFactory for FileOutputFactory { config: &TransportConfig, _endpoint_name: &str, _fault_tolerant: bool, - ) -> AnyResult>> { + ) -> AnyResult> { match config { - TransportConfig::FileOutput(config) => { - Ok(Some(Box::new(FileOutputEndpoint::new(config)?))) - } - _ => Ok(None), + TransportConfig::FileOutput(config) => Ok(Box::new(FileOutputEndpoint::new(config)?)), + _ => unexpected_output_transport_config("FileOutputFactory", config), } } } @@ -332,16 +330,13 @@ impl OutputTransportEndpointFactory for KafkaOutputFactory { config: &TransportConfig, endpoint_name: &str, fault_tolerant: bool, - ) -> AnyResult>> { + ) -> AnyResult> { match config { TransportConfig::KafkaOutput(config) => match fault_tolerant { - false => Ok(Some(Box::new(KafkaOutputEndpoint::new( - config, - endpoint_name, - )?))), - true => Ok(Some(Box::new(KafkaFtOutputEndpoint::new(config)?))), + false => Ok(Box::new(KafkaOutputEndpoint::new(config, endpoint_name)?)), + true => Ok(Box::new(KafkaFtOutputEndpoint::new(config)?)), }, - _ => Ok(None), + _ => unexpected_output_transport_config("KafkaOutputFactory", config), } } } @@ -356,12 +351,10 @@ impl OutputTransportEndpointFactory for RedisOutputFactory { config: &TransportConfig, _endpoint_name: &str, _fault_tolerant: bool, - ) -> AnyResult>> { + ) -> AnyResult> { match config { - TransportConfig::RedisOutput(config) => { - Ok(Some(Box::new(RedisOutputEndpoint::new(config)?))) - } - _ => Ok(None), + TransportConfig::RedisOutput(config) => Ok(Box::new(RedisOutputEndpoint::new(config)?)), + _ => unexpected_output_transport_config("RedisOutputFactory", config), } } } @@ -374,10 +367,10 @@ impl OutputTransportEndpointFactory for NullOutputFactory { config: &TransportConfig, _endpoint_name: &str, _fault_tolerant: bool, - ) -> AnyResult>> { + ) -> AnyResult> { match config { - TransportConfig::NullOutput => Ok(Some(Box::new(NullOutputEndpoint))), - _ => Ok(None), + TransportConfig::NullOutput => Ok(Box::new(NullOutputEndpoint)), + _ => unexpected_output_transport_config("NullOutputFactory", config), } } } @@ -385,8 +378,10 @@ impl OutputTransportEndpointFactory for NullOutputFactory { /// Creates an input transport endpoint instance using an input transport /// configuration, resolving secrets by reading `secrets_dir`. /// -/// Returns an error if there is a invalid configuration for the endpoint. -/// Returns `None` if the transport configuration variant is incompatible with an input endpoint. +/// Returns an error if there is an invalid configuration for the endpoint, or if the input +/// transport is handled by the transport registry but no factory is registered for it. +/// Returns `None` if the transport configuration variant is incompatible with an input endpoint, +/// including integrated output connectors. #[allow(unused_variables)] pub fn input_transport_config_to_endpoint( config: &TransportConfig, @@ -394,7 +389,14 @@ pub fn input_transport_config_to_endpoint( secrets_dir: &Path, ) -> AnyResult>> { let config = resolve_secret_references_via_json(secrets_dir, config)?; - builtin_input_transport_registry().create_endpoint(&config) + let endpoint = builtin_input_transport_registry().create_endpoint(&config)?; + if endpoint.is_none() && input_transport_uses_registry(&config) { + return Err(anyhow!( + "input transport factory for '{}' is not registered", + config.name() + )); + } + Ok(endpoint) } /// Creates an output transport endpoint instance using an output transport @@ -404,8 +406,10 @@ pub fn input_transport_config_to_endpoint( /// fault-tolerant output endpoint (but it will still return a non-FT endpoint /// if that's all it can do). /// -/// Returns an error if there is a invalid configuration for the endpoint. -/// Returns `None` if the transport configuration variant is incompatible with an output endpoint. +/// Returns an error if there is an invalid configuration for the endpoint, or if the output +/// transport is handled by the transport registry but no factory is registered for it. +/// Returns `None` if the transport configuration variant is incompatible with an output endpoint, +/// including integrated input connectors. #[allow(unused_variables)] pub fn output_transport_config_to_endpoint( config: &TransportConfig, @@ -414,7 +418,18 @@ pub fn output_transport_config_to_endpoint( secrets_dir: &Path, ) -> AnyResult>> { let config = resolve_secret_references_via_json(secrets_dir, config)?; - builtin_output_transport_registry().create_endpoint(&config, endpoint_name, fault_tolerant) + let endpoint = builtin_output_transport_registry().create_endpoint( + &config, + endpoint_name, + fault_tolerant, + )?; + if endpoint.is_none() && output_transport_uses_registry(&config) { + return Err(anyhow!( + "output transport factory for '{}' is not registered", + config.name() + )); + } + Ok(endpoint) } #[cfg(test)] @@ -479,6 +494,86 @@ mod tests { ); } + #[cfg(not(feature = "with-kafka"))] + #[test] + fn compiled_out_kafka_input_returns_missing_factory_error() { + let secrets_dir = tempfile::tempdir().unwrap(); + let config = serde_json::from_value(serde_json::json!({ + "name": "kafka_input", + "config": { + "topic": "topic", + "bootstrap.servers": "localhost:9092" + } + })) + .unwrap(); + + let error = match input_transport_config_to_endpoint(&config, "kafka", secrets_dir.path()) { + Err(error) => error, + Ok(_) => panic!("compiled-out kafka input should fail"), + }; + + assert!( + error + .to_string() + .contains("input transport factory for 'kafka_input' is not registered") + ); + } + + #[cfg(not(feature = "with-kafka"))] + #[test] + fn compiled_out_kafka_output_returns_missing_factory_error() { + let secrets_dir = tempfile::tempdir().unwrap(); + let config = serde_json::from_value(serde_json::json!({ + "name": "kafka_output", + "config": { + "topic": "topic", + "bootstrap.servers": "localhost:9092" + } + })) + .unwrap(); + + let error = + match output_transport_config_to_endpoint(&config, "kafka", true, secrets_dir.path()) { + Err(error) => error, + Ok(_) => panic!("compiled-out kafka output should fail"), + }; + + assert!( + error + .to_string() + .contains("output transport factory for 'kafka_output' is not registered") + ); + } + + #[cfg(not(feature = "with-redis"))] + #[test] + fn compiled_out_redis_output_returns_missing_factory_error() { + let secrets_dir = tempfile::tempdir().unwrap(); + let config = serde_json::from_value(serde_json::json!({ + "name": "redis_output", + "config": { + "connection_string": "redis://localhost:6379" + } + })) + .unwrap(); + + let error = match output_transport_config_to_endpoint( + &config, + "redis", + false, + secrets_dir.path(), + ) { + Err(error) => error, + Ok(_) => panic!("compiled-out redis output should fail"), + }; + + assert!( + error + .to_string() + .contains("output transport factory for 'redis_output' is not registered") + ); + } + #[test] fn explicit_transport_registries_dispatch_by_transport_name() { let mut input_registry = InputTransportRegistry::new(); @@ -512,6 +607,45 @@ mod tests { ); } + #[test] + fn transport_registry_direction_helpers_identify_fallback_transports() { + assert!(input_transport_uses_registry(&TransportConfig::EmptyInput)); + assert!(!input_transport_uses_registry(&TransportConfig::NullOutput)); + + assert!(output_transport_uses_registry(&TransportConfig::NullOutput)); + assert!(!output_transport_uses_registry( + &TransportConfig::EmptyInput + )); + } + + #[test] + fn selected_factory_rejects_mismatched_transport_config() { + let mut input_registry = InputTransportRegistry::new(); + input_registry.register(TransportConfig::EMPTY_INPUT, Box::new(FileInputFactory)); + let input_error = match input_registry.create_endpoint(&TransportConfig::EmptyInput) { + Err(error) => error, + Ok(_) => panic!("mismatched input factory should fail"), + }; + assert!( + input_error + .to_string() + .contains("FileInputFactory cannot create input endpoint") + ); + + let mut output_registry = OutputTransportRegistry::new(); + output_registry.register(TransportConfig::NULL_OUTPUT, Box::new(FileOutputFactory)); + let output_error = + match output_registry.create_endpoint(&TransportConfig::NullOutput, "null", true) { + Err(error) => error, + Ok(_) => panic!("mismatched output factory should fail"), + }; + assert!( + output_error + .to_string() + .contains("FileOutputFactory cannot create output endpoint") + ); + } + #[test] fn builtin_transport_registries_include_compiled_transport_names() { let input_registry = builtin_input_transport_registry();