connectors: make transport configs factory parsed#6529
Conversation
d2082e3 to
dc3c74c
Compare
|
small note on the registry key change: imo it belongs in this slice, not a later one. with transport config becoming built-ins behave the same. they still register string constants, and lookups still use borrowed |
4086edf to
dc3c74c
Compare
mythical-fred
left a comment
There was a problem hiding this comment.
This is the real prize from the #6446 thread: TransportConfig is now { name: Cow<'static, str>, config: JsonValue }, factories own their parsing, feldera-types no longer has to know every transport at compile time, and the SQL-side connector validation still catches the obvious wrong-direction mistakes (null_output on an input relation, datagen on an output relation) while letting genuinely unknown names through. The backward-compatibility tests (transport_config_deserializes_existing_config_payload_shape, transport_config_deserializes_existing_unit_variant_shape, transport_config_serializes_unit_transports_without_config_field) pin the migration: existing {name, config} payloads still deserialize, no-config transports may omit config, and serialization skips config: null. +1063/-340 even after touching 20 files is a reasonable size for this kind of architectural extraction.
A few real points:
known_transport_directionvssql_connector_transport_directiondivergence (crates/pipeline-manager/src/db/types/program.rs). Two near-identical tables with one delta: the SQL-side rejectsHTTP_INPUT/ADHOC_INPUT/CLOCK/HTTP_OUTPUTas "runtime-only" whileknown_transport_directionaccepts them. That's exactly right semantically (you can't declare an ad-hoc connector in SQLCONNECTORS), but the twomatches are easy to drift. A test asserts the SQL-rejection list today (generate_program_info_rejects_runtime_only_input_transports), but the relationship between the two functions —sql_connector_transport_direction(name).is_some()impliesknown_transport_direction(name).is_some()and the direction agrees — isn't pinned. A one-line property-style test that walks everyTransportConfig::*constant would catch a future addition that forgets one of the tables. The maintenance hazard is real because the canonical list of names lives infeldera-types::config::TransportConfigand the SQL filter lives inpipeline-manager, so adding a new transport now requires three coordinated edits (constant + registry + SQL filter) instead of two.- The boolean logic
sql_direction == Some(Output) || (known_transport_direction(name).is_some() && sql_direction.is_none())is right but harder to read than it needs to be. The intent is "reject if the transport is known and is not a valid input transport for SQL declarations." A small helper —fn rejects_as_input(name) -> bool— would compress the two call sites into a single boolean and document the rule. registered: BTreeMap<&'static str, ...>carryover from #6522. Same comment as on #6522: a third-party / dynamic factory can't easily inject a non-'staticname today. With the{ name: Cow<'static, str>, config }shape now stable, this is the last impedance-mismatch piece. Recommend widening toString(or at leastCow<'static, str>) before this lands so the dynamic-name path that motivates moving config parsing into factories is actually usable. Reading your own issue comment "small note on the registry key change: imo it belongs in this slice, not a later one" — agreed, fold it in.TransportConfig::is_transient()still hard-codes the runtime-only list (ADHOC_INPUT | HTTP_INPUT | HTTP_OUTPUT | CLOCK). That's the third copy of the same partition —is_transient,known_transport_direction,sql_connector_transport_direction— and they'll drift over time. Consider centralizing as aTransportClass { Input, Output, Transient }lookup table onTransportConfigitself, thenis_transient/ direction queries derive from it. Out of scope for this PR if you want, but worth filing a follow-up.integrated.rsremovals look right at a glance, and the open-codedmatchfor integrated connectors stays exactly where you'd expect (they still need schema/controller-handle wiring that's distinct from the factory shape). Worth a one-line comment inintegrated.rsnoting that integrated connectors are deliberately not in the registry because their factories need richer args.secret_resolver.rs-19 lines. The simplification (using the opaque JSON shape) is nice. One thing to double-check: every transport-specific secret-resolution rule that lived in the oldmatcharms is now handled in the factory or by the generic JSON walker, right? A regression here is silent (secret reference left unresolved → connector gets a literal${secret:...}string and fails opaquely at runtime). Worth one integration test per direction (input + output) that uses a${secret:...}reference and asserts the resolved value reaches the factory.
Direction is right. Direct the registry-key widening here and merge the two slices into a coherent story rather than carrying it into a third PR.
mythical-fred
left a comment
There was a problem hiding this comment.
Every blocker from my earlier pass is addressed cleanly:
- Drift guard between the two transport tables (
Guard SQL transport direction drift): the newrejects_sql_connector_transport(name, expected_direction)helper centralizes the boolean logic at both call sites — same intent ("reject if the transport is known and is not a valid SQL connector in this direction"), now one source of truth. Plusbuiltin_transport_names()lists all 24 built-in transports and the newknown_transport_direction_covers_*/sql_connector_transport_direction_is_subset_of_knownproperty tests pin (a)known_transport_directioncovers everyTransportConfig::*constant and (b)sql_connector_transport_direction(name).is_some()impliesknown_transport_direction(name).is_some()with matching direction. Adding a new transport now fails CI at the table walk, not silently at runtime. - Registry key widening (folded into #6522):
BTreeMap<String, ...>on both registries,register(impl Into<String>, ...), no behavior change for builtins. - Secret-resolution regression test (
Test transport secret resolution before parsing): two new tests (input_transport_secrets_are_resolved_before_factory_parsing,output_transport_secrets_are_resolved_before_factory_parsing) register a custom factory that assertsconfig.config["path"]equals the resolved value, then route a${secret:kubernetes:transport/...}reference through*_transport_config_to_endpoint_with_registry. Catches exactly the silent-failure mode I worried about (factory receives the unresolved${secret:...}literal). Nice extraction of the*_with_registryshim so tests don't need to mutate the global builtin registry. - Integrated connector doc (
Document integrated transport registry boundary): two-line comment oncreate_integrated_output_endpointsaying integrated connectors stay out of the registry because they need schema + controller lifecycle state at construction time. Future readers will not wonder.
is_transient() centralization is correctly deferred — I called it out as out-of-scope and the comment thread agrees. Worth a follow-up issue when you have a moment.
LGTM.
348e24e to
8f99d9d
Compare
refs #6446
imo this is the second slice after #6522. normal connector configs are open
{ name, config }values now, and the built-in transport factories parse their own payloads instead of makingfeldera-typesknow every transport.i kept the old json shape lgtm: existing
{ name, config }configs still deserialize, no-config transports can omitconfig, and sql property generation still catches known wrong-direction built-ins while letting unknown dynamic names through. idk if integrated connector registries belong in this stack, since they need a different factory shape around schemas, controller handles, and runtime state.