connectors: add transport endpoint registries#6522
Conversation
mythical-fred
left a comment
There was a problem hiding this comment.
Right shape for the first slice. The new InputTransportEndpointFactory / OutputTransportEndpointFactory traits + per-transport *InputFactory structs decompose the previous monolithic match in controller.rs::connect_input / connect_output cleanly. builtin_*_transport_registry() keeps every built-in registration in one place, and the Option<Box<dyn ...>> return convention preserves the "fall through to integrated connector" path correctly. Behavior is unchanged: the registry lookup hits the same factories the old match arms used to instantiate, error paths still go through ControllerError::input_transport_error / output_transport_error, secret resolution still happens up front. Net -89 lines of dispatch despite the new trait surface is a good sign.
A few thoughts:
registered: BTreeMap<&'static str, ...>inInputTransportRegistry. This is fine for built-ins (where keys areTransportConfig::FILE_INPUTetc.&'static strconstants), but it boxes any future third-party / user-provided factory into needing a'staticname. Dnreikronos's follow-up comment on #6529 ("the registry key change... belongs in this slice") is exactly right — switching toStringhere costs nothing at the call sites inbuiltin_*_transport_registry()(.to_string()) and unblocks any dynamic registration later. It's a one-line change in the type definition; doing it here saves a churn-the-API-again later. Suggest folding the #6529 key-type widening into this PR.get(&self, name: &str):BTreeMap<&'static str, _>::get(&str)needsname.to_owned()or aBorrow<str>workaround at the call site — the current code does.get(&config.name())which is&Stringagainst&&'static str. This compiles only becausename()returns&'static str, but a future caller passing an ownedStringwill get a confusing type error. Same root cause as the previous bullet.- Factories implement
create()by matching one specificTransportConfigvariant. That's the right shape, but it means the factory registry and theTransportConfigenum stay coupled: adding a new transport requires both editing the enum and adding a factory. The next slice (#6529) addresses this by moving config parsing into the factories, which is the real prize. Worth a one-line code comment atInputTransportEndpointFactorynoting that the current shape is transitional and that #6529 generalizes it — readers picking up either PR in isolation will be confused otherwise. KafkaInputFactoryalways createsKafkaFtInputEndpoint. I read this twice; the previous code did the same thing (the non-FT path was already dead). Carries forward, not a regression — but worth a note in the commit body thatKafkaInputandKafkaFtInputconfigs both go throughKafkaFtInputEndpointso future readers don't expect a non-FT branch here.- Test coverage. The PR is a pure refactor (no behavior change), and every transport is covered by existing integration tests, so the refactor is implicitly tested. No new tests needed. (No "tests for refactor" demand from me on this one — that's the right call for a no-op extraction.)
- Cosmetic. Several factory
struct *Factory;definitions are followed by a blank line thenimpl. Consistent. The#[cfg(feature = "...")]attributes on each factory + its impl are paired correctly. A possible compression: a smallfactory!macro to generate the per-transport boilerplate, but only worth it once a few more transports land; not worth doing now.
Direction is right; happy to APPROVE after the &'static str → String widening (or your call to defer it explicitly).
mythical-fred
left a comment
There was a problem hiding this comment.
The &'static str → String widening landed in Allow dynamic transport registry keys (adapterlib/src/transport.rs): BTreeMap<String, ...> on both registries, register(name: impl Into<String>, ...), and get(name: &str) reads cleanly (config.name().as_str()). The builtin &'static str constants flow in for free via Into, and a future dynamic factory can register an owned String without ceremony. The two ergonomic loose ends I flagged are gone.
LGTM.
7f58b45 to
2e5be66
Compare
| &self, | ||
| config: &TransportConfig, | ||
| ) -> AnyResult<Option<Box<dyn TransportInputEndpoint>>> { | ||
| let Some(factory) = self.get(config.name().as_str()) else { |
There was a problem hiding this comment.
I think this should return some error to help users diagnose why the registry cannot be found
There was a problem hiding this comment.
yep, handled in 3f43c20. i kept the raw registry returning none when the key is missing bc imo we still need that for the integrated connector fallback. the controller/helper path checks if the transport is one of the registry ones, and only then returns the missing-factory error.
There was a problem hiding this comment.
This should be documented
| fault_tolerant: bool, | ||
| ) -> AnyResult<Option<Box<dyn OutputEndpoint>>> { | ||
| let Some(factory) = self.get(config.name().as_str()) else { | ||
| return Ok(None); |
There was a problem hiding this comment.
same fix on the output side. the low-level registry can still say none for fallback, but the path users hit now gives an actual error when a registry-backed output transport has no factory. lgtm to me this way.
| TransportConfig::FileInput(config) => { | ||
| Ok(Some(Box::new(FileInputEndpoint::new(config)))) | ||
| } | ||
| _ => Ok(None), |
There was a problem hiding this comment.
shouldn't all these Ok() branches be in fact unreachable!()?
There was a problem hiding this comment.
yeah, fair. i changed create() so the selected factory returns the endpoint directly, no option. for the mismatch case i went with an explicit error instead of unreachable. imo that is nicer if someone wires a factory under the wrong key, bc they get a useful error instead of a panic.
| ) -> AnyResult<Option<Box<dyn TransportInputEndpoint>>> { | ||
| match config { | ||
| TransportConfig::FileInput(config) => { | ||
| Ok(Some(Box::new(FileInputEndpoint::new(config)))) |
There was a problem hiding this comment.
How come this can never return an error?
What if the config is malformed?
There was a problem hiding this comment.
It seems that new() in fact never fails, but I find that strange.
There was a problem hiding this comment.
yeah, that one is a bit weird at first glance. fileinputendpoint::new cannot fail right now, but i kept the trait fallible bc the next step moves more config parsing into factories. imo changing it to infallible now would just make us change it back later.
There was a problem hiding this comment.
malformed config can still fail where the endpoint constructor validates it. those factories still use ? and return the original error. file input is just one of the cases where new() does not fail today.
mythical-fred
left a comment
There was a problem hiding this comment.
Fourth commit tightens the registry contract nicely. Two changes worth calling out:
InputTransportEndpointFactory::create/OutputTransportEndpointFactory::createnow returnResult<Box<...>>(notResult<Option<Box<...>>>). Right call — a factory that has been dispatched to by transport name should neverOk(None). Theunexpected_input_transport_config/unexpected_output_transport_confighelpers convert the "wrong variant" case into a loud error with the factory name and transport name, so any dispatch bug is visible at runtime instead of silently short-circuiting. Theselected_factory_rejects_mismatched_transport_configtest pins this behaviour.- New
input_transport_uses_registry/output_transport_uses_registrypredicates enumerate everyTransportConfigvariant that goes through the registry. When the built-in registry doesn't return a factory (feature compiled out), the top-level helpers now returnUnknownInputTransport/UnknownOutputTransportrather than the previousOk(None)fall-through. Three#[cfg(not(feature = ...))]tests confirm the compiled-out kafka/redis paths surface the missing-factory error. This is the right shape — compiling out kafka should fail loudly on a kafka pipeline config, not produce a mysterious null endpoint.
Diff is +245/-111 on transport.rs, +30/-12 on controller.rs, +6/-7 on adapterlib/transport.rs; the churn is almost entirely mechanical Option<Box<...>> → Box<...>. No functional regressions on the happy path.
Non-blocking nits:
- The two
_uses_registryhelpers are hand-maintained enumerations ofTransportConfigvariants. If someone adds a new variant and forgets these, the compiled-out case will silently regress toOk(None). Consider a#[non_exhaustive]-style compile-time check, or at minimum a comment onTransportConfigreminding contributors to update the predicates. unexpected_input_transport_config/unexpected_output_transport_configreturnanyhow!(...)strings that get wrapped inControllerError::input_transport_error. That path is fine, but this class of failure is a programming bug (registry dispatched to the wrong factory), not a config error — adebug_assert!alongside the error would help surface it in tests without changing production behaviour.
Verdict stays APPROVE. Nothing here that should block merge.
mihaibudiu
left a comment
There was a problem hiding this comment.
This looks fine to me, but I am not the expert on connectors to make the final decision.
| &self, | ||
| config: &TransportConfig, | ||
| ) -> AnyResult<Option<Box<dyn TransportInputEndpoint>>> { | ||
| let Some(factory) = self.get(config.name().as_str()) else { |
There was a problem hiding this comment.
This should be documented
| &self, | ||
| config: &TransportConfig, | ||
| endpoint_name: &str, | ||
| fault_tolerant: bool, |
There was a problem hiding this comment.
Should this take the enum or just a boolean?
I don't know the answer
refs #6446
imo this is the small first slice from the issue thread. connector endpoint construction still accepts the existing transportconfig enum, but the normal input/output path now goes through catalog-owned registries and built-in factories instead of the big transport matches.
to fully ship the issue after this, idk that more should be piled in here. next step should be changing the public transport config shape closer to { name, config }, moving config parsing into connector factories, and then deciding whether inventory is useful for statically linked built-ins. lgtm to me as the first slice because it keeps behavior stable while making the next api change smaller.