Fixes #34479: skip stale-pipeline deletion when Data 360 listing is incomplete

[detail-app]

Detail bug report: View on Detail

Describe your changes:

Fixes #34479

I worked on the Salesforce Data 360 pipeline metadata connector because a failure in any of its three independent listing calls (get_pipelines_list chained _get_datastreams / _get_calculated_insights / _get_datatransforms via bare yield from) aborted the remaining listings. The topology runner swallows that producer exception and still runs the root node's post_process — mark_pipelines_as_deleted — unconditionally, so deletion reconciled against the partial pipeline_source_state and soft-deleted pipelines whose type was never fetched (worst case: the first listing fails, pipeline_source_state is empty, and all pipelines under the service are soft-deleted). markDeletedPipelines defaults to true, so this triggers on a normal metadata run.

The fix mirrors the mitigation two sibling connectors already use for this exact topology-runner behavior:

• A _pipeline_listing_complete flag (initialized/ reset to True at the start of every get_pipelines_list run) tracks whether all listings completed.
• get_pipelines_list isolates each listing in its own try/except: on failure it records a StackTraceError via status.failed, clears the flag, and continues to the next listing instead of aborting the chain.
• Data360PipelineSource now overrides mark_pipelines_as_deleted to skip deletion entirely when the listing was incomplete (logging a warning); otherwise it delegates to super(). This is the same pattern as Airflow's _dag_listing_complete (airflow/metadata.py), applied because — like Airflow — this connector deletes by service scope, so a global skip is correct (the Data 360 database sibling uses a per-scope skip since it deletes per schema).

Type of change:

• Bug fix

High-level design:

N/A — small bug fix (2 files). The approach tracks an already-shipped mitigation pattern (Airflow _dag_listing_complete), so no new architecture is introduced.

Tests:

Use cases covered

• A failure in the first listing (datastreams) does not soft-delete pipelines of any type, even though pipeline_source_state is empty.
• A failure in the middle listing (calculated insights) no longer skips the transforms listing — both unlisted types are preserved.
• A listing that yields some pipelines then raises still skips deletion while upserting the pipelines yielded before the failure.
• A successful run resets the incomplete flag and re-enables normal stale-pipeline cleanup.
• The happy path (all listings succeed) still deletes genuinely-stale pipelines.

Unit tests

• ingestion/tests/unit/topology/pipeline/test_data360pipeline.py — added TestData360PipelineListingIsolation and TestData360PipelineMarkDeleted (7 tests) covering listing isolation, the incomplete flag (set on failure, reset on the next run, set on a partial yield), and the mark_pipelines_as_deleted override's skip/delegate branches, plus an end-to-end test that drives get_pipeline → yield_pipeline → mark_pipelines_as_deleted in topology-runner order with a first-listing failure.
• make unit_ingestion / pytest -c ingestion/pyproject.toml ingestion/tests/unit/topology/pipeline/test_data360pipeline.py → 28 passed (21 existing + 7 new). ruff check, ruff format, the ruff-suppressions ratchet, and basedpyright (0 errors) all pass.

Backend integration tests

• Not applicable (no backend API changes).

Ingestion integration tests

• No connector integration test was added: the Data360Pipeline connector is unreleased (introduced in 138ba96b / PR New feature: Add Salesforce Data360 database and pipeline connectors #31918, present in no release tag), so no published server image knows the Data360Pipeline connection type to run one against. End-to-end behavior was instead verified manually against a real server (below).

Playwright (UI) tests

• Not applicable (no UI changes).

Manual testing performed

1. Brought up a real OpenMetadata 2.0.3 server via docker/docker-compose-quickstart/docker-compose.yml (mysql + elasticsearch + migrations + server); logged in as admin.
2. Pre-created a pipeline service (serviceType=CustomPipeline, since the released server doesn't know the unreleased Data360Pipeline type) and four pipelines (ds1, ci1, dt1, stale_ds), all deleted=false. The deletion path under test scopes by service name, so the connection type is irrelevant to it.
3. Mocked only the three Salesforce listing calls to inject failures, then drove the real Data360PipelineSource producer → yield_pipeline stage → mark_pipelines_as_deleted post-process against the live server, spying on the real metadata.delete_stale_entities (which calls DELETE /pipelines/deleteStale) and reading server-side deleted flags via /pipelines/name/<fqn>?include=all.
4. Confirmed:
   • First listing fails → deleteStale is not called; all four pipelines stay deleted=false (pre-fix, ds1 and stale_ds would have been soft-deleted).
   • All listings succeed → deleteStale is called once with the full live set; the genuinely-stale stale_ds is soft-deleted (deleted=true), the live ones stay deleted=false.
   • Partial listing (yield ds1 then raise) → deleteStale is not called; stale_ds stays deleted=false, while ds1 was listed before the failure.

UI screen recording / screenshots:

Not applicable.

Checklist:

• I have read the CONTRIBUTING document.
• My PR title is Fixes <issue-number>: <short explanation>
• My PR is linked to a GitHub issue via Fixes #<issue-number> above.
• I have commented on my code, particularly in hard-to-understand areas.
• For JSON Schema changes: I updated the migration scripts or explained why it is not needed. (N/A — no schema changes)
• For UI changes: I attached a screen recording and/or screenshots above. (N/A — no UI changes)
• I have added tests (unit / integration / Playwright as applicable) and listed them above.

• I have added a test that covers the exact scenario we are fixing.

---

Automatic Fixes PRs can be configured here.

[github-actions]

Hi there 👋 Thanks for your contribution!

The OpenMetadata team will review the PR shortly! Once it has been labeled as safe to test, the CI workflows
will start executing and we'll be able to make sure everything is working as expected.

Let us know if you need any help!

[github-actions]

✅ Playwright Results — workflow succeeded

Validated commit 34712bc965a8adab1fe98e2361e4f599487ff2b7 in Playwright run 36970830637, attempt 1.

✅ 105 passed · ❌ 0 failed · 🟡 0 flaky · ⏭️ 0 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) 1h 44m 2s

⏱️ Max setup 4m 48s · max shard execution 15m 10s · max shard-job elapsed before upload 21m 25s · reporting 2s

🌐 248.59 requests/attempt · 1.82 app boots/UI scenario · 0.00% common-shard skew

Optimization targets still in progress:

• Browser traffic was 248.59 requests per attempt (convergence target: fewer than 200).
• Application boot ratio was 1.82 per UI scenario (207 boots / 114 scenarios; convergence target: at most 1).

Shard	Passed	Failed	Flaky	Skipped	Lifecycle failed	Lifecycle flaky
✅ Shard chromium-01	40	0	0	0	0	0
✅ Shard ingestion-01	31	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

[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