Reject ingestion pipeline sourceConfig without type - #29566
Conversation
github-actions
Bot
added
Ingestion
safe to test
![gitar-bot[bot]](https://avatars.githubusercontent.com/in/827041?s=60&v=4){kind=small}
|
The Java checkstyle failed. Please run You can install the pre-commit hooks with |
❌ 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 |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
|
Pushed a fix for the current migration review thread in |
651fe14 to
a6e25e2
Compare
|
Want your agent to iterate on Greptile's feedback? Try greploops. |
✅ TypeScript Types Auto-UpdatedThe generated TypeScript types have been automatically updated based on JSON schema changes in this PR. |
|
|
Follow-up on the latest Gitar review summary: it reflects the superseded broad implementation. Current head The cited |
|
Follow-up for the review rerun on rebased head 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. |
| public void prepare(IngestionPipeline ingestionPipeline, boolean update) { | ||
| var service = getCachedParentOrLoad(ingestionPipeline.getService(), "", Include.NON_DELETED); | ||
| ingestionPipeline.setService(service.getEntityReference()); | ||
| validateSourceConfigHasType(ingestionPipeline); |
There was a problem hiding this comment.
[P2] Preserve self-registration for defaulted source config types
BaseWorkflow.get_or_create_ingestion_pipeline() forwards self.config.source.sourceConfig, while the shared Pydantic serializer defaults to exclude_unset=True. A YAML workflow that relies on the generated default discriminator therefore sends sourceConfig.config without type, and this new validation returns 400 during self-registration. The workflow then loses ingestion-pipeline status/progress tracking. Materialize the discriminator before the request (or coordinate the producer fix) and add a regression test for YAML that omits the defaulted type.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.
Suppressed comments (2)
ingestion/src/metadata/workflow/base.py:382
- The PR description states there are “no Python serialization changes” and that producer-side handling for
BaseWorkflowself-registration is deferred, but this change adds_source_config_with_explicit_type()and wires it intoget_or_create_ingestion_pipeline()to force the discriminator into the emitted payload. Please update the PR description (and linked issue resolution narrative) to match the actual scope, or revert this Python-side behavior if it truly must be deferred.
def _source_config_with_explicit_type(self) -> SourceConfig:
source_config = self.config.source.sourceConfig
config = source_config.config
if not isinstance(config, OpenMetadataBaseModel):
return source_config
config_type = getattr(config, "type", None)
if config_type is None:
return source_config
return source_config.model_copy(update={"config": config.model_copy(update={"type": config_type})})
openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java:1687
repairLegacySourceConfig()mutates the in-memory pipeline to add an inferredsourceConfig.config.type(and is invoked on everydeployIngestionPipeline()call). This appears to be an automatic legacy repair path, and in the REST deploy flow the mutated entity is persisted viacreateOrUpdate(...)after a successful deploy (see IngestionPipelineResource#deployPipelineInternal). That conflicts with the PR description’s stated scope of “no automatic legacy repairs / existing invalid rows require operational DB repair only”. Consider either (a) dropping this repair step and consistently rejecting legacy invalid configs, or (b) keeping the repair but updating the PR description and ensuring the inferred type is not persisted implicitly (e.g., repair a copy used only for deployment).
public PipelineServiceClientResponse deployIngestionPipeline(
IngestionPipeline ingestionPipeline, ServiceEntityInterface service) {
repairLegacySourceConfig(ingestionPipeline);
validateSourceConfigHasType(ingestionPipeline);
applyStreamableLogsConfig(ingestionPipeline);
Preserve listener-first registration while preventing a raced snapshot from replaying a live update.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 13 out of 13 changed files in this pull request and generated no new comments.
Suppressed comments (5)
openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java:1726
- The PR description says there are no database migrations and no automatic legacy repairs, but this method now performs an in-memory legacy repair (
repairLegacySourceConfig) before deploy, and the PR also adds a native DB backfill underbootstrap/sql/migrations/native/2.1.1/. Please either update the PR description/scope to include these legacy-repair + migration changes, or drop them to match the stated contract.
public PipelineServiceClientResponse deployIngestionPipeline(
IngestionPipeline ingestionPipeline, ServiceEntityInterface service) {
repairLegacySourceConfig(ingestionPipeline);
validateSourceConfigHasType(ingestionPipeline);
openmetadata-service/src/test/java/org/openmetadata/service/migration/v211/IngestionPipelineSourceConfigTypeBackfillTest.java:375
runBackfillhard-codes an exact statement count (3), which will make this test fail as soon as the 2.1.1 migration grows (even if the backfill is still correct). It’s safer to assert a minimum count (or assert presence of the expected backfill statements) while still executing everything.
List<String> statements = backfillStatements(database);
assertEquals(
3, statements.size(), "Expected reverse, service, and pipeline-only backfill statements");
openmetadata-service/src/test/java/org/openmetadata/service/migration/v211/IngestionPipelineSourceConfigTypeBackfillTest.java:402
backfillStatements()filters post-DDL scripts by a few hard-coded substrings. If the migration ever adds more backfill statements (e.g., for new service types), this test will silently stop executing/validating them, giving a false sense of coverage. Prefer returning/executing the fullgetPostDDLScripts()list (or filtering only empty statements).
return migrationFile.getPostDDLScripts().stream()
.filter(
statement ->
statement.contains("ReverseIngestion")
|| statement.contains("service_parent")
openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java:1609
- This PR is scoped/described as enforcing
sourceConfig.config.typevalidation/backfill, but it also changes the ingestion progress SSE streaming behavior (streamProgressnow usesProgressStreamEventEmitter) and adds a dedicated test class for it. Please either document this behavioral change in the PR description or split it into a separate PR so the ingestion-pipeline validation change can be reviewed/rolled out independently.
This issue also appears on line 1723 of the same file.
public void streamProgress(String pipelineFQN, UUID runId, SseEventSink eventSink, Sse sse) {
ProgressStreamEventEmitter emitter = new ProgressStreamEventEmitter(eventSink, sse);
Consumer<ProgressUpdate> listener = emitter::emitLiveUpdate;
Runnable onClose =
() -> progressTracker.unregisterProgressListener(pipelineFQN, runId, listener);
if (ProgressSseManager.getInstance().register(eventSink, sse, onClose)) {
progressTracker.registerProgressListener(pipelineFQN, runId, listener);
ProgressUpdate snapshot = getLatestProgressUpdate(pipelineFQN, runId);
if (snapshot != null) {
emitter.emitSnapshot(snapshot);
ingestion/src/metadata/workflow/base.py:382
- PR description states that producer-side handling for Python
BaseWorkflowself-registration is intentionally deferred, but this change adds_source_config_with_explicit_type()and wires it intoget_or_create_ingestion_pipeline(), which is exactly that producer-side fix. Please update the PR description/scope to reflect that Python behavior is now included (or drop this change if it truly belongs in a follow-up).
def _source_config_with_explicit_type(self) -> SourceConfig:
source_config = self.config.source.sourceConfig
config = source_config.config
if not isinstance(config, OpenMetadataBaseModel):
return source_config
config_type = getattr(config, "type", None)
if config_type is None:
return source_config
return source_config.model_copy(update={"config": config.model_copy(update={"type": config_type})})
Code Review ✅ Approved 4 resolved / 4 findingsRequires an explicit ✅ 4 resolved✅ Bug: Backfill added to already-released v1131 migration may never run
✅ Edge Case: Backfill no-ops on scalar sourceConfig.config rows, leaving them typeless
✅ Edge Case: Backfill only types databaseService metadata pipelines
✅ Edge Case: Snapshot emitted after listener registration can duplicate a live event
OptionsDisplay: 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 |
Fixes #28818.
Summary
This PR makes
sourceConfig.config.typea reliable discriminator for ingestion pipelines across API writes, legacy deployment handling, and Python workflow self-registration. It also closes an initial progress-stream race that could emit the same update twice.What changed
sourceConfig.config.typewith HTTP 400. Non-object configs are also rejected. Valid non-blank string discriminators remain accepted without normalization or inference.BaseWorkflowexplicitly includes Pydantic-defaulted source-config discriminators during self-registration, soexclude_unset=Truecannot omit the requiredtypefield.Scope and legacy data
This PR intentionally includes no database migration or data backfill. Existing persisted ingestion pipelines are not changed by normal reads or deployment. The deployment-time repair is in memory only; legacy records that cannot be identified unambiguously must be corrected explicitly before they can be deployed or updated.
No schema or generated-client changes are included.
Validation
mvn -pl openmetadata-service -Dtest=IngestionPipelineRepositoryTest,IngestionPipelineProgressStreamTest,IngestionProgressTrackerTest,ProgressSseManagerTest,ServiceProgressStreamerTest test— 89 passed.pytest -c ingestion/pyproject.toml ingestion/tests/unit/workflow/test_base_workflow.py— 15 passed.mvn -pl openmetadata-service spotless:check— passed.git diff --check origin/main— passed.