Reject ingestion pipeline sourceConfig without type

[ayush-shah]

Fixes #28818.

Summary

Prevent future invalid ingestion-pipeline writes by requiring sourceConfig.config to be an object with a usable type discriminator at repository persistence time. Validation runs from IngestionPipelineRepository.prepare() after service resolution and therefore applies to POST, PUT/upsert, and PATCH.

Behavior

• Missing sourceConfig, missing/null config, or a missing/null/non-string/blank type returns HTTP 400 with sourceConfig.config.type is required.
• Scalar, list, or otherwise non-object configs return HTTP 400 with sourceConfig.config must be an object with type.
• Raw maps with a non-blank string type and generated Java config objects carrying their discriminator remain valid.
• Validation is presence-only: unknown non-blank discriminator strings remain accepted. The server does not infer, default, normalize, or mutate the value.
• An unrelated PATCH against an already-invalid legacy row is rejected until the discriminator is supplied or repaired operationally.

Scope

This PR prevents new invalid writes only. The JSON schema and generated clients are unchanged, and there are no Python serialization changes, database migrations, or automatic legacy repairs. Existing affected installations require an independent operational database repair; reads and persisted legacy data are not modified by this change.

Python BaseWorkflow self-registration may receive HTTP 400 when its serializer omits a defaulted discriminator. Producer-side handling is intentionally deferred to a separate follow-up.

Changes

• Add the focused repository validation helper and negative/positive unit matrix.
• Align the Playwright PipelineClass fixture with the runtime requirement.
• Update the stale Collate integration assertion separately in open-metadata/openmetadata-collate#5273.

Validation

• IngestionPipelineRepositoryTest: 30 passed.
• mvn -pl openmetadata-service -Dskip.npm -Dskip.yarn -DskipDocker spotless:check: passed.
• PipelineExecution.spec.ts: 4 passed, including setup and teardown.
• git diff --check: passed.
• Combined OpenMetadata + companion Collate cross-repository workflow: passed, including the Collate Maven build and MySQL/Elasticsearch integration suite.
• Full Playwright TypeScript checking was attempted; current main reports unrelated existing errors, with no error in the changed fixture.

Local REST persistence reproduction

The same invalid POST payload used sourceConfig.config: {}.

Server	HTTP result	Direct MySQL verification
Pre-fix revision 6b1a2e8d	201 Created	One invalid ingestion-pipeline row persisted
PR head 73c3e589	400 Bad Request: sourceConfig.config.type is required	Zero rows persisted

The fixed-side request was repeated with fresh service and pipeline names and again returned HTTP 400 with zero persisted rows.

Greptile Summary

This PR adds server-side validation in IngestionPipelineRepository.prepare() to reject ingestion pipeline writes whose sourceConfig.config is missing, non-object, or lacks a non-blank type discriminator, returning HTTP 400 in all such cases. Existing persisted data and reads are unaffected; no migrations or schema changes are included.

• validateSourceConfigHasType (static, package-private) handles three config shapes: raw Map (reads type directly), generated Java objects (converts via JsonUtils.getMap / Jackson, then reads type), and non-object types like scalars or lists (Jackson throws IllegalArgumentException, mapped to a distinct 400 message).
• A ten-case parameterized negative test matrix plus two positive acceptance tests covers null configs, empty maps, blank strings, wrong types, enums in raw maps, scalars, and lists.
• The Playwright PipelineClass fixture is updated to supply type: 'PipelineMetadata' so E2E tests pass the new validation.

Confidence Score: 5/5

Safe to merge — the change is additive validation only, with no data mutation, no schema changes, and no side effects on existing valid pipelines.

The validation logic correctly handles all documented input shapes, the error messages are consistent with the test assertions, and the Playwright fixture is aligned with the new requirement. The only untested branch is the Enum path inside the generated-config arm, which is harmless dead code under standard Jackson configuration.

Files Needing Attention: No files require special attention.

Important Files Changed

Filename	Overview
openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java	Adds validateSourceConfigHasType static helper in prepare() to reject pipelines missing a non-blank sourceConfig.config.type; handles raw Maps, generated Java objects, and non-object types correctly.
openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/IngestionPipelineRepositoryTest.java	Adds a comprehensive parameterized negative test matrix (10 invalid cases) and two positive acceptance tests for the new validation; all cases are well-chosen and correctly assert the expected error messages.
openmetadata-ui/src/main/resources/ui/playwright/support/entity/PipelineClass.ts	Fixture updated to supply type: 'PipelineMetadata' in sourceConfig.config, aligning the E2E test payload with the new server-side validation requirement.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A[POST / PUT / PATCH IngestionPipeline] --> B[EntityRepository.prepareInternal]
    B --> C[IngestionPipelineRepository.prepare]
    C --> D[getCachedParentOrLoad - resolve service]
    D --> E[validateSourceConfigHasType]

    E --> F{sourceConfig or config is null?}
    F -- yes --> G[400: sourceConfig.config.type is required]

    F -- no --> H{config instanceof Map?}
    H -- yes raw Map --> I[get type from Map directly]
    H -- no generated object --> J[JsonUtils.getMap convert to Map via Jackson]

    J --> K{IllegalArgumentException? scalar or list}
    K -- yes --> L[400: sourceConfig.config must be an object with type]

    I --> M{type is non-blank String?}
    J --> M

    M -- yes --> N[Valid - proceed]
    M -- no --> O{generatedConfig AND type instanceof Enum?}
    O -- yes --> N
    O -- no --> G

Loading

_{Reviews (14): Last reviewed commit: "ingestion: reject source configs without..." | Re-trigger Greptile}

[github-actions]

The Java checkstyle failed.

Please run mvn spotless:apply in the root of your repository and commit the changes to this PR.
You can also use pre-commit to automate the Java code formatting.

You can install the pre-commit hooks with make install_test precommit_install.

[github-actions]

❌ PR checklist incomplete

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

• Issue Bug: Ingestion workflow redeployment fails due to missing sourceConfig.config.type in automated pipeline creation #28818 is missing Shipping → Status.
• Issue Bug: Ingestion workflow redeployment fails due to missing sourceConfig.config.type in automated pipeline creation #28818 is missing Shipping → Domain.
• Issue Bug: Ingestion workflow redeployment fails due to missing sourceConfig.config.type in automated pipeline creation #28818 is missing Shipping → Release.

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.

[ayush-shah]

Pushed a fix for the current migration review thread in 080ec95044: the source-config backfill now uses an explicit service-to-config-type mapping join instead of CASE expressions, so unmatched rows are not written with JSON null.\n\nValidation run locally:\n- git diff --check\n- rtk mvn -pl openmetadata-service -DskipTests=false -Dskip.npm -Dskip.yarn -DskipDocker -Dlicense.skip=true -Dcheckstyle.skip=true -Dspotless.check.skip=true -Dtest=org.openmetadata.service.migration.utils.v200.MigrationUtilTest#backfillsMetadataSourceConfigTypesWithMySqlJsonSet,org.openmetadata.service.migration.utils.v200.MigrationUtilTest#backfillsMetadataSourceConfigTypesWithPostgresJsonbSet test\n- rtk mvn -pl openmetadata-service -DskipTests -Dskip.npm -Dskip.yarn -DskipDocker -Dlicense.skip=true spotless:check

[ayush-shah]

Fresh author summary for current head 651fe14d8b5: the branch now enforces sourceConfig.config.type, moves the backfill to the v2.0 migration, maps metadata service relationships to their config discriminators, preserves generated Python workflow discriminator values, and updates Java/Python/Playwright coverage. Current status is not merge-ready yet: PR metadata validation is failing and several CI jobs are still in progress, so this remains blocked until those checks settle.

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

Copilot reviewed 9 out of 9 changed files in this pull request and generated 1 comment.

[greptile-apps]

Want your agent to iterate on Greptile's feedback? Try greploops.

[Copilot]

Pull request overview

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

[github-actions]

✅ TypeScript Types Auto-Updated

The generated TypeScript types have been automatically updated based on JSON schema changes in this PR.

[Copilot]

Pull request overview

Copilot reviewed 11 out of 15 changed files in this pull request and generated 1 comment.

[Copilot]

Pull request overview

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

[sonarqubecloud]

Quality Gate passed for 'open-metadata-ingestion'

Issues
0 New issues
0 Accepted issues

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

See analysis details on SonarQube Cloud

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[ayush-shah]

Follow-up on the latest Gitar review summary: it reflects the superseded broad implementation. Current head 45260c629f contains only the repository validation, focused tests, and Playwright fixture alignment; it has no Python serialization, schema/codegen, migration, legacy-repair, or progress-streaming changes.

The cited streamProgress listener-before-snapshot ordering is unchanged from the current main base. A concurrent update can indeed be emitted live and then repeated by the snapshot, but moving the snapshot before listener registration can lose an update in the opposite race window. If this pre-existing edge case is addressed, it should use atomic registration/snapshot semantics or sequence-aware de-duplication in a separate change. No progress-streaming change is included in this PR.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[gitar-bot]

Code Review 👍 Approved with suggestions 3 resolved / 4 findings

Enforces sourceConfig.config.type presence in ingestion pipelines through backend validation, Python serialization fixes, and an idempotent v200 data backfill. Please address the potential event duplication issue identified during the listener registration refactor.

💡 Edge Case: Snapshot emitted after listener registration can duplicate a live event

📄 openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java:1290-1303 📄 openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java:1316-1322

In the refactored IngestionPipelineRepository.streamProgress, the progress listener is registered before the current snapshot is read and emitted:

progressTracker.registerProgressListener(pipelineFQN, runId, listener);
ProgressUpdate snapshot = getLatestProgressUpdate(pipelineFQN, runId);
if (snapshot != null) {
  emitProgressUpdate(eventSink, sse, snapshot);
}

The previous implementation emitted the snapshot before registering the listener. With the new ordering, if a live ProgressUpdate fires in the window between registerProgressListener and getLatestProgressUpdate, the client receives that update via the listener and then receives the snapshot (which reflects the same or newer state), producing a duplicate initial event. For progress rendering this is benign, but if a terminal (PIPELINE_COMPLETE/ERROR) update arrives via the listener in that window it will close the sink, and the subsequent snapshot emit becomes a no-op via the isClosed() guard — so correctness is preserved, only a possible duplicate remains. Consider reading the snapshot before registering the listener, or de-duplicating by tracking the last emitted update, to avoid the redundant event.

✅ 3 resolved

✅ Bug: Backfill added to already-released v1131 migration may never run

📄 openmetadata-service/src/main/java/org/openmetadata/service/migration/mysql/v1131/Migration.java:26 📄 openmetadata-service/src/main/java/org/openmetadata/service/migration/postgres/v1131/Migration.java:26 📄 openmetadata-service/src/main/java/org/openmetadata/service/migration/utils/v1131/MigrationUtil.java:111-125
backfillDatabaseMetadataSourceConfigType is wired into mysql/v1131/Migration and postgres/v1131/Migration. On main the migration tree already contains much newer versions (up through v200 / 2.0.0). OpenMetadata's MigrationWorkflow/MigrationProcessImpl records applied versions in SERVER_CHANGE_LOG and skips runDataMigration() for any version already processed. Any deployment that has already upgraded past 1.13.1 (i.e. nearly every target user of this fix) will never re-execute v1131's runDataMigration(), so the backfill of the 'known persisted bad database metadata pipeline shape' will silently not run for them. Only brand-new installs upgrading through v1131 would get it. The backfill should be placed in the current in-development/unreleased migration version (e.g. v200 / 2.0.0) so it actually executes for existing deployments. Please confirm which version is the active upgrade target before merging.

✅ Edge Case: Backfill no-ops on scalar sourceConfig.config rows, leaving them typeless

📄 openmetadata-service/src/main/java/org/openmetadata/service/migration/utils/v1131/MigrationUtil.java:121-124 📄 openmetadata-service/src/main/java/org/openmetadata/service/migration/utils/v1131/MigrationUtil.java:126-135 📄 openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java:298-312
The backfill WHERE clause only guards sourceConfig.config IS NOT NULL (MySQL JSON_EXTRACT(... '$.sourceConfig.config') IS NOT NULL, Postgres i.json #> '{sourceConfig,config}' IS NOT NULL). If a persisted row stored sourceConfig.config as a scalar (e.g. the string "DatabaseMetadata") rather than an object, the row still matches the predicate, but JSON_SET(..., '$.sourceConfig.config.type', ...) / jsonb_set(..., '{sourceConfig,config,type}', ...) cannot add a member to a scalar and effectively no-ops (MySQL returns the document unchanged; Postgres jsonb_set on a non-object path also yields no member). Those rows therefore remain without a usable type, yet the new validateSourceConfigHasType will reject any subsequent create/update of them with HTTP 400. If the scalar shape is among the 'known bad' shapes this PR intends to fix, it is not covered. Consider detecting/handling the scalar-config case (or asserting it never occurs) and add a test for it.

✅ Edge Case: Backfill only types databaseService metadata pipelines

📄 openmetadata-service/src/main/java/org/openmetadata/service/migration/utils/v200/MigrationUtil.java:103-117
backfillDatabaseMetadataSourceConfigType only sets sourceConfig.config.type = 'DatabaseMetadata' for pipelines joined via er.fromEntity = 'databaseService' with pipelineType = 'metadata'. Metadata pipelines for other service types (dashboardService, messagingService, etc.) that were also persisted without a sourceConfig.config.type discriminator are not backfilled. If the new repository-level validation (described in the PR summary but not part of this delta) rejects any untyped sourceConfig.config across all service types, those existing pipelines would start failing on their next create/update without a corresponding backfill. If the intent is genuinely database-only, this is fine; otherwise consider broadening the backfill (or adding parallel statements) to cover the other service-type metadata config discriminators. Flagging as minor since the validation code is not in this diff and cannot be verified here.

🤖 Prompt for agents

Code Review: Enforces `sourceConfig.config.type` presence in ingestion pipelines through backend validation, Python serialization fixes, and an idempotent v200 data backfill. Please address the potential event duplication issue identified during the listener registration refactor.

1. 💡 Edge Case: Snapshot emitted after listener registration can duplicate a live event
   Files: openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java:1290-1303, openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java:1316-1322

   In the refactored `IngestionPipelineRepository.streamProgress`, the progress listener is registered *before* the current snapshot is read and emitted:
   
   ```java
   progressTracker.registerProgressListener(pipelineFQN, runId, listener);
   ProgressUpdate snapshot = getLatestProgressUpdate(pipelineFQN, runId);
   if (snapshot != null) {
     emitProgressUpdate(eventSink, sse, snapshot);
   }
   ```
   
   The previous implementation emitted the snapshot *before* registering the listener. With the new ordering, if a live `ProgressUpdate` fires in the window between `registerProgressListener` and `getLatestProgressUpdate`, the client receives that update via the listener and then receives the snapshot (which reflects the same or newer state), producing a duplicate initial event. For progress rendering this is benign, but if a terminal (PIPELINE_COMPLETE/ERROR) update arrives via the listener in that window it will close the sink, and the subsequent snapshot emit becomes a no-op via the `isClosed()` guard — so correctness is preserved, only a possible duplicate remains. Consider reading the snapshot before registering the listener, or de-duplicating by tracking the last emitted update, to avoid the redundant event.

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}

[ayush-shah]

Follow-up for the review rerun on rebased head 73c3e589: this is the same three-file patch (stable patch ID unchanged), now based directly on current main. The PR contains no streamProgress or listener-registration change; that code is byte-for-byte unchanged from the merge base and is absent from this diff.

The duplicate-event race is plausible in the pre-existing implementation, but reading the snapshot first trades it for a lost-update race. A correct fix would need atomic snapshot/registration semantics or sequence-aware de-duplication and belongs in a separate progress-streaming change. No progress-streaming modification is applicable to this PR.