Fixes #31152: emit Snowflake Sink lineage in the KafkaConnect connector - #31153
Fixes #31152: emit Snowflake Sink lineage in the KafkaConnect connector#31153ulixius9 wants to merge 16 commits into
Conversation
…gistry Delete _parse_datasets_from_config (now DefaultResolver.resolve_datasets) and route the sink branch of _match_topic_to_dataset plus the dataset-parsing call site in yield_pipeline_lineage_details through the per-connector-class resolver registry. This is what makes managed Confluent Cloud Snowflake Sink connectors produce lineage datasets end-to-end; the CDC/Debezium branch is untouched.
…est idiom Code review findings on the resolver-registry wiring commit: - Restore self._resolver_for(pipeline_details) in _match_topic_to_dataset's sink branch (the brief's literal Step 5 code). The prior classname-dispatch workaround (KafkaconnectSource._resolver_for(self, ...)) silently broke subclass overrides of _resolver_for, including the real KafkaconnectSourceTests(KafkaconnectSource) subclass used in tests/unit/source/pipeline/test_kafkaconnect.py. Drop the comment that justified the workaround. - Fix the test side instead: replace None-as-self calls in test_kafkaconnect_snowflake_sink.py with the repo's established object.__new__(KafkaconnectSource) idiom via a small _new_source() helper, matching the pattern already used throughout tests/unit/topology/pipeline/test_kafkaconnect.py. No assertions changed. - Remove two stale comments left behind by the earlier _parse_datasets_from_config deletion, and reword the _match_topic_to_dataset docstring's sink-matching description to reflect resolver-based matching instead of name equality.
…ble FQNs Add Snowflake to the connector class -> service type / hostname key maps, gate get_dataset_entity's Priority 1 FQN and Priority 3 search string on fully_qualified so Snowflake sinks get a correct 4-part FQN while Debezium's logical-server-name CDC datasets keep their existing 3-part shape, and warn with the dbServiceNames remediation hint when a table can't be resolved.
… CDC FQN guard find_database_service_by_hostname only probed hostPort and host, neither of which SnowflakeConnection has, so SERVICE_TYPE_HOSTNAME_KEYS["Snowflake"] extracted snowflake.url.name and compared it against nothing: the Snowflake service was never matched and the four-part FQN path was unreachable. Probe account as well (hostPort/host still win, so other service types are unchanged) and tolerate a per-service-type domain suffix, with the .snowflakecomputing.com datum kept in constants.py. Covered by a test that drives a real DatabaseService + SnowflakeConnection end to end. Rewrite the CDC regression guard to assert on fqn.build's keyword arguments instead of a joined string, which was invariant under any permutation of the slots, and add the table.include.list shape (schema set, unqualified) that nothing guarded. Both mutations of the production gate now fail the guard. Also return early from Priority 3 when no table name is known, hoist the repeated get_db_service_names() call, and tidy the test module imports. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…shapes Appends tests encoding the column/table shapes measured against live Confluent Cloud and Snowflake on 2026-08-05: schematization creates one column per top-level Avro field, uppercased, and a nested record becomes a single VARIANT column instead of being flattened into child columns. Also locks in end-to-end dataset resolution for a real Confluent Cloud connector config with mixed valid/invalid Snowflake identifiers. No source changes.
Call build_column_lineage over the live-observed topic/table shapes with field and column FQNs populated, asserting the exact four edges, the single ADDRESS VARIANT target, and that RECORD_METADATA is never a target. Drop the vacuous test_record_metadata_produces_no_edge, narrow the extractor test to what it verifies, and guard the remaining vacuous assertions. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
SnowflakeSinkResolver.column_mappings now walks the topic's Avro schema and emits one source-field -> target-column pair per leaf when a Flatten SMT is in the transform chain, joining the field path with the transform's delimiter. Only a transform's own `type` is consulted: Confluent Cloud omits defaulted properties, so the absence of snowflake.enable.schematization says nothing about whether flattening happens. With no Flatten configured it still returns [], keeping the live-verified 1:1 name inference (and its single VARIANT column for a nested record) in charge. The explicit-mapping branch of build_column_lineage had never executed because nothing populated column_mappings; it resolved topic fields through get_column_fqn(table_entity=topic), which cannot work because Topic has no .columns. It now dispatches through _get_entity_column_fqn. _get_topic_field_fqn only searched three levels, so it could not reach a nested leaf: the Avro parser inserts a type-named level under every record-typed field, putting `street` at OrderEvent.address.Address.street. Add a breadth-first descent as a fallback after the existing explicit levels, leaving the Debezium after-over-before precedence untouched. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Same-named leaves collapsed onto one topic field. column_mappings discarded the path (source_column=path[-1]) and the topic-field lookup matched by bare name, so shipping.city and billing.city both resolved to whichever the search found first and BILLING_CITY was published with shipping.city as its upstream. A nested leaf shadowing a top-level field of the same name failed the same way. Before the deep search these resolved to None and were dropped; resolving them to the wrong field is worse than not resolving them at all. source_column now carries the dotted source path and _get_topic_field_fqn walks it, stepping over the type-named level the Avro parser inserts under every record-typed field. Dots are unambiguous because no Avro or JSON-schema field name may contain one, and a path miss falls through to the existing by-name searches, which need an exact full-string match. extract_column_mappings is the only other writer of source_column; it has no production caller and emits single-segment names, which remain valid one-segment paths. The by-name deep search is deleted rather than reordered: with path resolution it has no reachable purpose, and a bare deep name is inherently ambiguous, so any answer it gives may be wrong. That also restores CDC resolution to exactly the shipped code path, removing the risk that it reported a Debezium pre-image as a column's upstream. Kafka Connect's Flatten recurses into STRUCT only, so an array is copied through whole and an array of records is one VARIANT column. Descending into it invented columns that do not exist and suppressed the real one, since a non-empty mapping list turns off 1:1 inference for every column. Treat ARRAY as a leaf; MAP needs no guard and a nullable record is a STRUCT, so it still flattens. Adds the after-over-before coverage the legacy CDC fixture never provided (it leaves both children None), coverage for both halves of the call-site guard, and corrects the guard comment: skipping a missing topic is housekeeping, not crash avoidance, since the resolver tolerates None and answers []. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…nostics The NOT FOUND lineage summary hardcoded three CDC/JDBC-style config keys (database.hostname, database.server, connection.host) and never consulted SERVICE_TYPE_HOSTNAME_KEYS, so a managed Snowflake sink -- whose host lives under snowflake.url.name -- always reported "hostname: NOT SET" even though the connector had plainly declared one. Confirmed live against a Confluent Cloud connector reporting snowflake.url.name = " FMFAHQK-GI58232.snowflake computing.com". That misled support triage into thinking the connector never set a host. Add _debug_hostname(), which derives the service-type-specific key(s) from CONNECTOR_CLASS_TO_SERVICE_TYPE/SERVICE_TYPE_HOSTNAME_KEYS first, falls back to the three legacy keys, and reuses the existing _extract_hostname whitespace stripping. Diagnostic string only -- no resolution or control-flow change. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ver a superset
Three review must-fixes.
1. basedpyright baseline entries key on {startColumn, endColumn, lineCount}
and carry no line number, so growing the Priority-1 fqn.build() call from
8 lines to 12 orphaned its baselined reportArgumentType and failed CI's
static-checks gate. Hoist the two inline conditionals into locals, which
restores the baselined line count and reads better at the call site. Pure
extraction: the gate is still strictly `fully_qualified`.
2. SnowflakeSinkResolver narrowed what the key-list search used to answer, and
registering a connector class leaves no fallback: it ignored the
snowflake.database / snowflake.schema variations (both already listed in
constants) and produced nothing for topics.regex or a failed /topics fetch.
Read both key forms from the shared constants, defer to DefaultResolver
when no topic can be derived, and delegate match_topic for the datasets
that fallback produces, which carry no source_topic.
3. sinks/__init__.py's registration import was invisible to the tests -- the
test module imports sinks.snowflake directly -- so deleting it kept every
test green while production silently lost both Snowflake keys. Re-export
the module and assert the registration from a fresh interpreter that
imports only the package.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
❌ PR checklist incompleteThis PR cannot be merged until the following are addressed on its linked issue:
The fields live on the linked issue in the Shipping project (open the issue → right sidebar → Projects). After you set them, re-run this check (or push a commit) — issue/project changes do not re-trigger it automatically. Maintainers can bypass this check by adding the |
github-actions
Bot
added
Ingestion
safe to test
![gitar-bot[bot]](https://avatars.githubusercontent.com/in/827041?s=60&v=4){kind=small}
There was a problem hiding this comment.
Pull request overview
This PR fixes KafkaConnect lineage extraction for Snowflake Sink connectors by introducing a sink-specific dataset resolver registry and a Snowflake resolver that can (a) derive target tables from topics (including Snowflake’s sanitization/hash behavior), (b) preserve topic→table pairing for reliable matching, and (c) resolve Snowflake services by matching snowflake.url.name against Snowflake service account (with domain-suffix tolerance). It also wires resolver-provided column mappings (Flatten SMT) into column-level lineage.
Changes:
- Add
sinks/resolver registry +DefaultResolver(verbatim key-list logic) and introduceSnowflakeSinkResolverfor topic-derived table resolution and Flatten SMT mappings. - Extend dataset/service resolution: new dataset fields (
source_topic,fully_qualified), Snowflake service-type mappings, and hostname matching enhancements (probeaccount, strip known domain suffixes). - Add comprehensive unit tests covering managed/self-managed Snowflake sink configs, service discovery, dataset FQN construction, and column-level lineage (including Flatten SMT).
Reviewed changes
Copilot reviewed 8 out of 8 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/sinks/base.py | Adds resolver ABC + registry and moves the historical dataset key-list parsing into DefaultResolver. |
| ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/sinks/snowflake.py | Implements Snowflake-specific dataset derivation, topic matching, and Flatten SMT column mappings; registers resolver under both managed and self-managed class keys. |
| ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/sinks/init.py | Ensures resolver registration occurs via package import side effects and re-exports resolver utilities. |
| ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/models.py | Adds source_topic and fully_qualified to preserve topic↔table pairing and control 3-part vs 4-part FQN construction. |
| ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/constants.py | Adds Snowflake connector↔service-type mapping, Snowflake hostname key, and service-connection host probing/suffix stripping data. |
| ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/metadata.py | Delegates sink dataset resolution to the registry, improves Snowflake service matching, fixes explicit column mapping handling for Topic entities, and enhances diagnostics. |
| ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/client.py | Simplifies connector config extraction to always return the flat config map from the Connect v1 response. |
| ingestion/tests/unit/topology/pipeline/test_kafkaconnect_snowflake_sink.py | Adds extensive test coverage for Snowflake sink dataset resolution, service discovery, FQN slotting, and column lineage (including Flatten SMT + CDC regression guards). |
Suppressed comments (1)
ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/metadata.py:583
- In
get_dataset_entity, Priority 2 (configureddbServiceNames) still builds the table FQN usingdatabase_name=dataset_details.databaseandschema_name=dataset_details.schema. For CDC/Debezium datasetsfully_qualifiedis False, sodatabaseis a logical server name and must remain in the schema slot (withdatabase_name=None) just like Priority 1 now does. As-is, CDC lineage can fail to resolve when hostname matching misses anddbServiceNamesis relied on (and it can also regress cases liketable.include.list=inventory.orderswhereschemais set but should not override the logical server name).
# Priority 2: Use configured dbServiceNames
db_service_names = self.get_db_service_names()
for dbservicename in db_service_names or ["*"]:
dataset_entity = self.metadata.get_by_name(
entity=dataset_details.dataset_type,
fqn=fqn.build(
metadata=self.metadata,
entity_type=dataset_details.dataset_type,
table_name=dataset_details.table,
database_name=dataset_details.database,
schema_name=dataset_details.schema,
service_name=dbservicename,
),
)
| "connector.class": "SnowflakeSink", | ||
| "input.data.format": "AVRO", | ||
| "kafka.api.key": "IWZOEA4Q46ZJDE52", | ||
| "kafka.api.secret": "****************", | ||
| "kafka.auth.mode": "KAFKA_API_KEY", | ||
| "kafka.endpoint": "SASL_SSL://pkc-56d1g.eastus.azure.confluent.cloud:9092", |
…e datasets Address three review findings on the Snowflake Sink lineage work. Copilot: the captured Confluent Cloud fixture carried the maintainer's real Confluent API key, cluster hostname, Snowflake account, database, schema and user. Every identifier is now an obviously fake value of the same shape, while the properties the fixtures exist to pin survive untouched: the leading space Confluent stores in snowflake.url.name, secrets rendered as fixed-width masks rather than omitted, the absence of every defaulted property, and the exact key set. Assertions were retargeted to the anonymised values, so they assert the same behaviour. gitar-bot: fully_qualified was computed as `database and schema`, which conflated "are both parts present" with "is this dataset Snowflake-shaped". Priority 1 of get_dataset_entity uses the flag only to choose a slot, so a sink configuring snowflake.database.name alone had its real database pushed into the schema slot and built an FQN that can never match the table. A Snowflake sink's database is always a real database, never a Debezium-style logical server name, so the flag now follows `database` alone -- and a config naming only one of the two keys warns, naming the key that is missing. gitar-bot: datasets were built only for discovered/configured topics, so a topic named solely in snowflake.topic2table.map -- what a topics.regex subscription with partial discovery leaves behind -- produced no dataset and no log line. The map is explicit configuration pairing a concrete topic with a concrete table, so those entries now yield datasets too, appended after the discovered topics to preserve their order, and are logged. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
| topic_names = self._topic_names(config, topics) | ||
| if not topic_names: | ||
| # A connector can subscribe by topics.regex, and get_connector_topics answers | ||
| # None on any transport failure, so an empty topic list is not proof that the | ||
| # connector writes nothing: topic2table.map still names its tables. Deferring | ||
| # keeps self-managed sinks at the lineage they had before this resolver existed. | ||
| logger.info( | ||
| f"Snowflake sink '{config.get('name')}' declares no topics; " | ||
| f"resolving its target from the connector config keys instead" | ||
| ) | ||
| datasets = DefaultResolver().resolve_datasets(config, topics) | ||
| if not datasets: | ||
| logger.warning(f"Snowflake sink '{config.get('name')}' declares no topics; no lineage can be built") | ||
| return datasets |
There was a problem hiding this comment.
💡 Edge Case: topic2table.map lost when no topics are discovered at all
_with_mapped_topics recovers map-only topics only on the non-empty topic_names path. When a topics.regex sink discovers zero concrete topics (or get_connector_topics fails and returns None), resolve_datasets takes the early-return branch and delegates to DefaultResolver, which does not parse snowflake.topic2table.map — so a Snowflake sink whose tables are named solely by the map loses all lineage, the near-miss of the very case this commit fixes. Consider seeding datasets from the map in the empty-topic branch too (e.g. run _with_mapped_topics(config, [], mapping) before falling back to DefaultResolver).
Was this helpful? React with 👍 / 👎
Code Review 👍 Approved with suggestions 2 resolved / 3 findingsAdds Snowflake sink lineage and table FQN resolution to the KafkaConnect connector using a new modular sink resolver registry. Consider ensuring 💡 Edge Case: topic2table.map lost when no topics are discovered at all📄 ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/sinks/snowflake.py:87-100 📄 ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/sinks/snowflake.py:119 📄 ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/sinks/snowflake.py:233-247
✅ 2 resolved✅ Edge Case: Partial Snowflake db/schema config places database in schema slot
✅ Edge Case: topic2table.map entries for undiscovered topics are silently dropped
🤖 Prompt for agentsOptionsDisplay: compact → Showing less information. Comment with these commands to change the behavior for this request:
Was this helpful? React with 👍 / 👎 | Powered by Gitar — free for open source |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 8 out of 8 changed files in this pull request and generated no new comments.
Suppressed comments (2)
ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/sinks/snowflake.py:105
- When no topics are discoverable (e.g., topics.regex subscription or topic fetch failure), this resolver falls back to DefaultResolver even if
snowflake.topic2table.mapis present. That loses the topic↔table pairing (source_topicstays None), somatch_topic()can only succeed when the topic name happens to equal the table name — which defeats the purpose of topic2table.map and can still yield no lineage even though the map fully specifies the targets.
topic_names = self._topic_names(config, topics)
if not topic_names:
# A connector can subscribe by topics.regex, and get_connector_topics answers
# None on any transport failure, so an empty topic list is not proof that the
# connector writes nothing: topic2table.map still names its tables. Deferring
ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/metadata.py:606
- In the Priority-3 ES fallback search, fully-qualified datasets currently search only by
schema.table, ignoringdatabaseeven when it is present. For Table wildcard search this widens the query from*.db.schema.tableto*.*.schema.table, which increases the chance of resolving the wrong table across multiple databases/services (and emitting incorrect lineage) when service resolution fails.
# Build search string: schema.table format (with proper quoting for special chars)
search_parts = [
part
for part in (
dataset_details.schema if dataset_details.fully_qualified else dataset_details.database,
dataset_details.table,
)
if part
]
search_string = ".".join(fqn.quote_name(part) for part in search_parts)
|
✅ Playwright Results — workflow succeededValidated commit ✅ 108 passed · ❌ 0 failed · 🟡 0 flaky · ⏭️ 0 skipped · 🧰 0 lifecycle flaky PerformanceBlocking targets: ✅ met · Optimization targets: 🟡 in progress Shard-job maxima below are not the full workflow wall time; the linked run includes build, fixture, planning, and reporting. 🕒 Full workflow signal wall (to summary) 50m 8s ⏱️ Max setup 3m 38s · max shard execution 10m 47s · max shard-job elapsed before upload 19m 10s · reporting 3s 🌐 209.11 requests/attempt · 1.76 app boots/UI scenario · 0.00% common-shard skew Optimization targets still in progress:
How to debug locally# Download playwright-test-results-<shard> artifact and unzip
npx playwright show-trace path/to/trace.zip # view trace |
Describe your changes:
Fixes #31152
I worked on making the
kafkaconnectconnector emitTopic → Pipeline → Snowflake Tablelineage, with column-level detail, for Snowflake Sink connectors — because today it emits nothing in every realistic configuration. Two defects stacked: the Snowflake Sink derives its table from the topic name, so the generic key-list parser matched no table key and produced zero datasets (and whensnowflake.topic2table.mapwas set, the topic side of each pair was discarded and pairing degraded to positional index plus name equality, which fails by construction); and the Snowflake service could not be resolved at all, because there was no service-type mapping and hostname matching probes onlyhostPort/host, whichSnowflakeConnectiondoes not have. Rather than add another branch to an already# noqa: C901function, I introduced a connector-class-keyed sink resolver registry, moved the existing key-list logic into aDefaultResolververbatim, and gave Snowflake its own resolver.Type of change:
High-level design:
Architecture. New
ingestion/src/metadata/ingestion/source/pipeline/kafkaconnect/sinks/package:base.py—SinkDatasetResolverABC (resolve_datasets/match_topic/column_mappings), asink_resolver_registrybuilt on the repo's sharedenum_register()primitive,get_resolver(), andDefaultResolver.snowflake.py—java_string_hashcode(),snowflake_table_name(), andSnowflakeSinkResolver, registered under bothSnowflakeSink(the Confluent Cloud managed short plugin name) andSnowflakeSinkConnector(self-managed). Registering only one silently falls back to the default resolver with no error.metadata.pynow delegates instead of branching inline, so it shrinks. This follows the Factory/Registry pattern already documented indocs/design-patterns.md, and keeps connector-specific logic out of shared code —metadata.pycontains zero occurrences of the string "snowflake"; all Snowflake knowledge lives inconstants.pyandsinks/.DefaultResolveris a verbatim move, which is what makes this safe. Debezium/CDC, JDBC, and S3/GCS/Azure storage sinks all depend on that logic. It was differentially fuzzed against the original across 6,701 cases (6,365resolve_datasetsconfigs + 336match_topiccases) with zero divergence in return values or logger call sequences. The pre-existing 1,586-linetest_kafkaconnect.pypasses untouched at every commit.Two model fields carry the fix (
models.py, hand-written Pydantic — not schema-generated, so nomake generateis owed):source_topic— the topic↔table pairing was previously thrown away at parse time and guessed back positionally. Carrying it makesmatch_topican O(1) exact lookup and deletes the guessing.fully_qualified— needed becauseget_dataset_entity's Priority 1 hardcodeddatabase_name=None, schema_name=dataset.database. That shape is CDC-specific (for Debezium,databaseholdstopic.prefix, a logical server name, not a real database). A naive generalisation such asdatabase if schema else Nonesilently breaks Debezium configured withtable.include.list = inventory.orders, so the resolver declares qualification explicitly instead and the 4-part FQN is built only for datasets that opt in.Service resolution adds
SERVICE_CONNECTION_HOST_ATTRIBUTESandSERVICE_TYPE_HOST_DOMAIN_SUFFIXESas data inconstants.py, so hostname matching can also probeaccountand strip a service-type-keyed domain suffix. This is what letssnowflake.url.namematch a real Snowflake service and makesdbServiceNamesgenuinely optional.Alternatives rejected. Inlining the Snowflake special case was cheaper but the generic key-list machinery cannot express "derive the table from
topicswhen no mapping key exists", and every sink added that way makes the next harder. A dedicated Confluent Cloud connector was rejected outright — it would fork ~2,800 lines of topic resolution, service matching, and lineage plumbing for one sink type, and Confluent Cloud is already a first-class path in the existing client.Backward compatibility. Both new model fields default to preserving current behaviour, and no existing construction site passes either.
DefaultResolverapplies to every connector class without a dedicated resolver, so non-Snowflake paths are unchanged.SnowflakeSinkResolveris a strict superset ofDefaultResolver: it accepts thesnowflake.database/snowflake.schemavariation keys and falls back toDefaultResolverwhen no topics are discoverable, so self-managed sinks cannot lose lineage they had.Deliberately out of scope: migrating the storage-sink branch into the registry (drags S3/GCS/Azure regression risk into a customer-facing fix), and
ReplaceFieldSMT support.Tests:
Use cases covered
snowflake.topic2table.map→ one dataset per topic, table name derived from the topicabs(javaStringHashCode(topic))(om-lineage-test→OM_LINEAGE_TEST_702890019)snowflake.topic2table.mapwith renaming, including a partial map where some topics are mapped and others derivesnowflake.url.namewithlineageInformation.dbServiceNamesleft emptyVARIANTcolumn;RECORD_METADATAcorrectly yields no edge)FlattenSMT configured → nested leaf paths mapped to flattened columns with the configured delimiterUnit tests
ingestion/tests/unit/topology/pipeline/test_kafkaconnect_snowflake_sink.py(1,667 lines)test_kafkaconnect_snowflake_sink.py,test_kafkaconnect.py,test_kafkaconnect_service_discovery.py, andtests/unit/source/pipeline/test_kafkaconnect.py.--cov=metadata.ingestion.source.pipeline.kafkaconnect):sinks/snowflake.py99%,sinks/base.py94%,constants.py100%,models.py96%,client.py65%,metadata.py77% (large pre-existing file), overall 79%.fully_qualifiedgate removed entirely, and a "column lineage" suite in which an earlyreturn Noneinsidebuild_column_lineagekilled zero tests. Assertions now targetfqn.build's keyword arguments and realColumnLineageedges.Backend integration tests
Ingestion integration tests
Covered by the unit suite above plus the live end-to-end run below. Fixtures are captured verbatim from a real Confluent Cloud
GET /connectors/{name}response and a realDESC TABLE, including their quirks (a leading space insnowflake.url.name, masked-not-omitted secrets, and defaulted properties absent from the response).Playwright (UI) tests
Manual testing performed
Verified end-to-end against live Confluent Cloud and Snowflake, not only unit tests:
addressrecord, and a hyphenated topic name) plus a managed Snowflake Sink withinput.data.format: AVRO,SNOWPIPE_STREAMING,snowflake.enable.schematization: true, and notopic2table.map.lineageInformation.dbServiceNamesleft empty.GET /api/v1/lineage/getLineagethat all three tables haveTopic → Pipeline → Tableedges with column-level detail:order_events_flat → ORDER_EVENTS_FLAT— 6 column edgesorder_events_nested → ORDER_EVENTS_NESTED— 4 column edges,address → ADDRESSas a singleVARIANTcolumnom-lineage-test → OM_LINEAGE_TEST_702890019— 4 column edges (the hash-derivation path)RECORD_METADATA, and that the Snowflake service resolved purely by matchingsnowflake.url.nameagainst the service'saccount.UI screen recording / screenshots:
Not applicable.
Checklist:
Fixes <issue-number>: <short explanation>Fixes #<issue-number>above.openmetadata-spec/, so no regeneration or migration is owed.