fix(ingestion): resolve topic sample values against unwrapped messages

[IceS2]

What

Auto Classification on a messaging service samples every field as null and therefore never applies a PII tag. Reproduced against a live Kafka source: the stored sample is literally

{"Customer": {"customer_id": null, "full_name": null, "email": null, "ssn": null, ...}}

with no tag on any field. This affects every Avro and JSON Schema topic, so PII detection on Kafka and Redpanda has not worked since it shipped in #28446.

Why

The schema parsers wrap a topic schema in a root RECORD named after the schema, so MessagingSampler._flatten_field builds column paths like Customer.email. The message on the wire is {"email": ...} with no such wrapper, so _resolve walked msg["Customer"]["email"] and returned None for every field.

The column names cannot change. AutoClassificationProcessor._find_column_by_dotted_path matches on exactly that dotted scheme — its docstring says so — and it is shared with database and storage entities, attaching the tag via column.fullyQualifiedName. Renaming the columns would break tagging for every entity type.

The fix is therefore confined to value resolution: try the full path first, then the path with the schema root stripped. A genuinely wrapped message keeps resolving as it does today.

Why it was not caught

The integration fixture hand-built a flat schemaFields list and pushed it straight into a CreateTopicRequest, bypassing the connector. That shape cannot come out of ingestion — yield_topic runs the registry schema through parse_avro_schema / parse_json_schema, both of which emit a root RECORD. With a flat list the lookup succeeded and the assertions passed.

The fixture now derives its fields from the real JSON schema parser, so it tracks whatever the parser produces, and the assertions walk the tree rather than the top level.

Also

SamplerProcessor dereferenced config.processor unconditionally. Messaging and storage auto-classification have nothing configurable there, so the block is routinely omitted, and the workflow died at startup with a bare AttributeError: 'NoneType' object has no attribute 'model_dump'. An empty inner config: hit ValidationError one line later. Both are now accepted.

The same unguarded line exists in ProfilerProcessor, but every profiler config supplies a processor in practice and there is no unit coverage there, so it is left alone rather than changed speculatively.

Scope

_resolve and _flatten_field are referenced only by MessagingSampler.fetch_sample_data. Its subclasses are Kafka, Redpanda (via Kafka), and the Kinesis and PubSub stubs. Database and storage samplers are separate classes and are untouched.

New unit tests cover the nested schema, the wrapped-message case, both shapes present at once, the flat schema, an absent field, and a topic with no schema fields. They fail on main and pass here.

[github-actions]

❌ PR checklist incomplete

This PR cannot be merged until the following are addressed on its linked issue:

• No GitHub issue is linked. Link an issue in the Development section of the PR (or add Fixes #12345 to the description). For a same-org cross-repo issue, add Fixes open-metadata/<repo>#123 to the description.

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 skip-pr-checks label.

[Copilot]

Pull request overview

This PR fixes messaging-topic auto-classification sampling by resolving field values against both the schema-wrapped dotted path (e.g., Customer.email) and the unwrapped on-the-wire message shape (e.g., {"email": ...}), and updates tests to reflect the real parsed schema shape produced by the JSON schema parser.

Changes:

• Update messaging sampler value resolution to try the full dotted path first, then a root-stripped variant for unwrapped messages.
• Make SamplerProcessor tolerate missing/empty processor config blocks (common for messaging/storage auto-classification).
• Strengthen unit/integration tests to use real schema parsing and to traverse nested schema trees when asserting tags.

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
ingestion/src/metadata/sampler/messaging/sampler.py	Adds root-tolerant value resolution for messaging samples (wrapped vs unwrapped messages).
ingestion/src/metadata/sampler/processor.py	Guards optional processor block / empty processor.config when building profiler_config.
ingestion/tests/unit/sampler/test_sampler_processor_config.py	New unit test ensuring SamplerProcessor construction works without processor config.
ingestion/tests/unit/sampler/test_messaging_sampler.py	New unit tests for dotted-path column naming and message value resolution across schema/message shapes.
ingestion/tests/integration/auto_classification/messaging/conftest.py	Integration fixture now derives schemaFields from the real JSON schema parser and adds helpers to traverse nested fields.
ingestion/tests/integration/auto_classification/messaging/test_kafka_classification.py	Assertions updated to find leaf fields in nested schemas and validate tags across all leaves.
ingestion/tests/integration/auto_classification/messaging/test_redpanda_classification.py	Updates fixture usage to include schema text and use shared nested-field lookup helper.

Suppressed comments (1)

ingestion/src/metadata/sampler/messaging/sampler.py:115

• _resolve treats a resolved value of None the same as a missing path, because _walk returns None for both cases. This can cause incorrect fallbacks: a genuinely wrapped message like { "Order": { "email": null }, "email": "unwrapped@example.com" } would resolve Order.email to the unwrapped value, even though the full path exists and should win (per the docstring). Use a sentinel to distinguish “missing” from “present but null”, and only fallback when the path is actually missing.

    def _walk(msg: dict, dotted: str) -> object:
        """Walk a dotted path into nested dicts, returning None when absent."""
        cur: object = msg
        for part in dotted.split("."):
            if isinstance(cur, dict):

[Copilot]

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated no new comments.

Suppressed comments (1)

ingestion/src/metadata/sampler/messaging/sampler.py:138

• _resolve currently falls back to the root-stripped path whenever the full dotted path is absent. This means that if a message is actually wrapped (root key exists) but the nested field is missing (e.g., {"Order": {}} plus a top-level email), the sampler will incorrectly pick the sibling top-level value for Order.email, which contradicts the intended "wrapped message keeps winning" behavior described in the docstring.

        for candidate in (dotted, dotted.split(".", 1)[-1]):
            value = MessagingSampler._walk(msg, candidate)
            # Only an absent path falls through: a field explicitly set to null is
            # the schema-correct answer and must not inherit a same-named sibling.
            if value is not MISSING:
                return value
        return None

[github-actions]

✅ Playwright Results — workflow succeeded

Validated commit db3a4d08f94fc3339b63c787e217c597add1834f in Playwright run 30857180004, attempt 1.

✅ 105 passed · ❌ 0 failed · 🟡 0 flaky · ⏭️ 3 skipped · 🧰 0 lifecycle flaky

Performance

Blocking 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) 47m 51s

⏱️ Max setup 3m 1s · max shard execution 10m 56s · max shard-job elapsed before upload 15m 58s · reporting 4s

🌐 212.07 requests/attempt · 1.78 app boots/UI scenario · 0.00% common-shard skew

Optimization targets still in progress:

• Browser traffic was 212.07 requests per attempt (convergence target: fewer than 200).
• Application boot ratio was 1.78 per UI scenario (206 boots / 116 scenarios; convergence target: at most 1).

Shard	Passed	Failed	Flaky	Skipped	Lifecycle failed	Lifecycle flaky
✅ Shard chromium-01	46	0	0	3	0	0
✅ Shard ingestion-01	25	0	0	0	0	0
✅ Shard ingestion-02	34	0	0	0	0	0

📦 Download artifacts

How to debug locally

# Download playwright-test-results-<shard> artifact and unzip
npx playwright show-trace path/to/trace.zip    # view trace

[Copilot]

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated no new comments.

Suppressed comments (2)

ingestion/src/metadata/sampler/messaging/sampler.py:138

• _resolve() always tries two candidates, but for flat schemas (no dot in dotted) the second candidate is identical to the first, causing redundant lookups for every column/message. This is a measurable (though small) performance regression for common flat-topic schemas.

        for candidate in (dotted, dotted.split(".", 1)[-1]):
            value = MessagingSampler._walk(msg, candidate)
            # Only an absent path falls through: a field explicitly set to null is
            # the schema-correct answer and must not inherit a same-named sibling.
            if value is not MISSING:
                return value
        return None

ingestion/tests/unit/sampler/test_messaging_sampler.py:70

• The helper _topic() is annotated as fields: list[FieldModel] but the test calls it with None (@pytest.mark.parametrize("fields", [[], None])). This makes the type hint internally inconsistent and can trip type-checkers/static analysis.

def _topic(fields: list[FieldModel]) -> Topic:
    return Topic(
        id=uuid.uuid4(),
        name="om_orders_avro",
        partitions=1,
        service=EntityReference(id=uuid.uuid4(), type="messagingService"),
        messageSchema=MessageSchema(schemaFields=fields),

[sonarqubecloud]

Quality Gate passed for 'open-metadata-ingestion'

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
100.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[gitar-bot]

Code Review ✅ Approved 1 resolved / 1 findings

Fixes Kafka and Redpanda PII auto-classification by updating topic sample value resolution to fall back against unwrapped messages and handles missing processor configurations, addressing the _resolve fallback conflates explicit null with missing field finding.

✅ 1 resolved

✅ Edge Case: _resolve fallback conflates explicit null with missing field

📄 ingestion/src/metadata/sampler/messaging/sampler.py:121-134
In _resolve, when the full path (Order.email) resolves to an explicit None in a genuinely wrapped message, the loop falls through to the root-stripped candidate (email) and may pick up an unrelated same-named top-level key. Example: {"email": "leak@x.com", "Order": {"email": None}} returns "leak@x.com" even though the schema-correct value is null. This is a rare message shape and unlikely in practice, but it means a null wrapped field can silently inherit a different value. If strict fidelity matters, distinguish 'present-but-null' from 'absent' by having _walk return a sentinel rather than None.

Options

Display: compact → Showing less information.

Comment with these commands to change the behavior for this request:

Compact

gitar display:verbose

_{Was this helpful? React with 👍 / 👎 | Gitar | Powered by Gitar — free for open source}