Fixes #20341: add automation Workflow retention to the Data Retention app - #32620
Conversation
… app The `automations_workflow` table grows without bound: reverse ingestion, test connection and query runner runs each leave a Workflow entity behind and nothing ever removes them. Reverse metadata made this visible by creating workflows in volume. Adds a `workflowRetentionPeriod` setting (default 30 days, 0 to retain forever) that the weekly Data Retention job drains through the existing batching, stats and failure-reporting path. Two deliberate choices: - The field is not `required`, so no upgrade migration is needed. jsonschema2pojo emits an initialized default and Jackson leaves absent keys alone, so a configuration saved before this change reads 30. - Deletion goes through the repository rather than a bulk SQL delete. A Workflow holds the service connection it ran against, so dropping the row alone would strand its secrets in an external secrets manager and orphan its owner relationship rows. Rows are selected by age alone rather than by terminal status: a workflow stuck in Pending or Running still leaks, and a 30-day-stale updatedAt already means the run is dead. Per-workflow failures are counted and skipped instead of rethrown, since batches are ordered oldest first and aborting on one bad row would wedge the cleanup permanently. Co-Authored-By: Claude Opus 5 <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
backend
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.
🟡 Changes recommended
The current batch-drain termination condition can stop workflow cleanup early on partial-delete failures, leaving additional expired workflows unprocessed.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
This PR extends the Data Retention application to prevent unbounded growth of automations_workflow by introducing a configurable retention period and deleting expired Workflow entities via the repository deletion path (so related secrets/relationships are cleaned up).
Changes:
- Added
workflowRetentionPeriod(default 30 days, 0 = retain forever) to the Data Retention app configuration schema and surfaced it in the UI app schema/help. - Implemented automation-workflow cleanup in
DataRetentionusing a DAO query to batch-select old workflow IDs and hard-delete them viaEntity.deleteEntity. - Added unit tests for defaulting behavior and batch-drain termination, plus an integration test covering end-to-end deletion of an old automation workflow.
File summaries
| File | Description |
|---|---|
| openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/dataRetention/DataRetention.java | Adds the automation-workflow cleanup step driven by the new retention config. |
| openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/WorkflowDocStoreDAOs.java | Adds DAO query to list workflow IDs older than a cutoff, for batch cleanup. |
| openmetadata-spec/src/main/resources/json/schema/entity/applications/configuration/internal/dataRetentionConfiguration.json | Adds the new workflowRetentionPeriod field to the internal Data Retention configuration schema. |
| openmetadata-service/src/main/resources/json/data/app/DataRetentionApplication.json | Updates the seeded Data Retention app config to include the new retention setting. |
| openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/dataRetention/DataRetentionTest.java | Adds unit tests for schema-default fallback and drain termination behavior. |
| openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/DataRetentionAppIT.java | Adds an IT that creates workflows, backdates one, triggers the app, and asserts old workflow deletion. |
| openmetadata-ui/src/main/resources/ui/src/jsons/applicationSchemas/DataRetentionApplication.json | Surfaces the new retention field in the UI-rendered app configuration schema. |
| openmetadata-ui/src/main/resources/ui/public/locales/en-US/Applications/DataRetentionApplication.md | Adds operator help text for the new field. |
| openmetadata-ui/src/main/resources/ui/src/generated/entity/applications/configuration/internal/dataRetentionConfiguration.ts | Regenerated TS type to include workflowRetentionPeriod. |
| openmetadata-ui/src/main/resources/ui/src/generated/metadataIngestion/workflow.ts | Regenerated TS type(s) to include workflowRetentionPeriod in CollateAIAppConfig. |
| openmetadata-ui/src/main/resources/ui/src/generated/metadataIngestion/applicationPipeline.ts | Regenerated TS type(s) to include workflowRetentionPeriod in CollateAIAppConfig. |
| openmetadata-ui/src/main/resources/ui/src/generated/metadataIngestion/application.ts | Regenerated TS type(s) to include workflowRetentionPeriod in CollateAIAppConfig. |
| openmetadata-ui/src/main/resources/ui/src/generated/entity/services/ingestionPipelines/serviceProgressEvent.ts | Regenerated TS type(s) to include workflowRetentionPeriod in CollateAIAppConfig. |
| openmetadata-ui/src/main/resources/ui/src/generated/entity/services/ingestionPipelines/ingestionPipeline.ts | Regenerated TS type(s) to include workflowRetentionPeriod in CollateAIAppConfig. |
| openmetadata-ui/src/main/resources/ui/src/generated/entity/applications/marketplace/createAppMarketPlaceDefinitionReq.ts | Regenerated TS type(s) to include workflowRetentionPeriod in CollateAIAppConfig. |
| openmetadata-ui/src/main/resources/ui/src/generated/entity/applications/marketplace/appMarketPlaceDefinition.ts | Regenerated TS type(s) to include workflowRetentionPeriod in CollateAIAppConfig. |
| openmetadata-ui/src/main/resources/ui/src/generated/entity/applications/app.ts | Regenerated TS type(s) to include workflowRetentionPeriod in CollateAIAppConfig. |
| openmetadata-ui/src/main/resources/ui/src/generated/api/services/ingestionPipelines/createIngestionPipeline.ts | Regenerated TS type(s) to include workflowRetentionPeriod in CollateAIAppConfig. |
Review details
- Files reviewed: 8/18 changed files
- Comments generated: 2
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| LOG.info("Initiating automation workflows cleanup: Retention = {} days.", retentionPeriod); | ||
| long cutoffMillis = getRetentionCutoffMillis(retentionPeriod); | ||
|
|
||
| executeWithStatsTracking("automation_workflows", () -> deleteExpiredWorkflows(cutoffMillis)); |
| @SqlQuery( | ||
| "SELECT id FROM automations_workflow WHERE updatedAt < :cutoffTs " | ||
| + "ORDER BY updatedAt LIMIT :limit") | ||
| List<String> listIdsBeforeCutoff(@Bind("cutoffTs") long cutoffTs, @Bind("limit") int limit); |
✅ Generated Sources Auto-UpdatedThe generated TypeScript types ( |
There was a problem hiding this comment.
🔵 Needs a closer look
The new automation-workflow cleanup uses a drain condition based on “deleted < batch size”, but per-entity delete failures make that signal unreliable and can prematurely stop draining while expired workflows remain.
Review details
Suppressed comments (1)
openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/dataRetention/DataRetention.java:493
executeWithStatsTrackingdrains batches based ondeleted < BATCH_SIZE, butdeleteExpiredWorkflowsreturns only the count of successful per-entity deletes. If any workflow in a batch fails deletion,deletedbecomes< BATCH_SIZEand the drain stops early even when there are still many expired workflows remaining (withBATCH_SIZE=10_000, this can leave a large backlog until future weekly runs). Use a drain condition based on "no progress" (e.g., stop only when a batch deletes 0) for this per-entity delete path.
private void cleanAutomationWorkflows(int retentionPeriod) {
LOG.info("Initiating automation workflows cleanup: Retention = {} days.", retentionPeriod);
long cutoffMillis = getRetentionCutoffMillis(retentionPeriod);
executeWithStatsTracking("automation_workflows", () -> deleteExpiredWorkflows(cutoffMillis));
- Files reviewed: 10/20 changed files
- Comments generated: 0 new
- Review effort level: Lite
There was a problem hiding this comment.
🔵 Needs a closer look
The automation workflow cleanup can stop after the first batch when any per-entity delete fails, leaving additional expired workflows unpruned.
Review details
Suppressed comments (1)
openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/dataRetention/DataRetention.java:493
- cleanAutomationWorkflows currently uses executeWithStatsTracking, which considers the table drained when a batch deletes fewer than BATCH_SIZE rows. Since deleteExpiredWorkflows deletes per ID and can skip failures, any single failed delete will make the batch delete count < BATCH_SIZE and prematurely stop the drain even when more expired workflows remain beyond the first batch.
private void cleanAutomationWorkflows(int retentionPeriod) {
LOG.info("Initiating automation workflows cleanup: Retention = {} days.", retentionPeriod);
long cutoffMillis = getRetentionCutoffMillis(retentionPeriod);
executeWithStatsTracking("automation_workflows", () -> deleteExpiredWorkflows(cutoffMillis));
- Files reviewed: 10/20 changed files
- Comments generated: 0 new
- Review effort level: Lite
Jest test CoverageUI tests summary
|
There was a problem hiding this comment.
🔵 Needs a closer look
The current drain termination condition for workflow cleanup can stop early after partial per-row delete failures, leaving eligible workflows unprocessed.
Review details
Suppressed comments (2)
openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/dataRetention/DataRetention.java:493
- The drain termination condition for automation workflows is currently based on the number successfully deleted (< BATCH_SIZE). If any deletion fails within a full batch (e.g., 9 deleted, 1 failed),
deleteExpiredWorkflowsreturns 9 andexecuteWithStatsTrackingwill treat the table as drained and stop, leaving many eligible workflows unprocessed. The drain should only stop when a batch deletes 0 (no progress), since per-row failures can make the deleted count smaller than the batch even when more rows remain.
private void cleanAutomationWorkflows(int retentionPeriod) {
LOG.info("Initiating automation workflows cleanup: Retention = {} days.", retentionPeriod);
long cutoffMillis = getRetentionCutoffMillis(retentionPeriod);
executeWithStatsTracking("automation_workflows", () -> deleteExpiredWorkflows(cutoffMillis));
openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/WorkflowDocStoreDAOs.java:234
- This query orders by
updatedAtand limits, butautomations_workflowhas no index on the generatedupdatedAt/updatedatcolumn in either MySQL or Postgres schemas (only PK(id) and unique(nameHash)). On large catalogs this can force a full scan + sort every batch of a weekly job, which can become operationally expensive precisely when the table has grown large. Consider adding an index on the generatedupdatedAt/updatedatcolumn (and validating with EXPLAIN) so the retention job can drain efficiently.
@SqlQuery(
"SELECT id FROM automations_workflow WHERE updatedAt < :cutoffTs "
+ "ORDER BY updatedAt LIMIT :limit")
List<String> listIdsBeforeCutoff(@Bind("cutoffTs") long cutoffTs, @Bind("limit") int limit);
- Files reviewed: 10/20 changed files
- Comments generated: 0 new
- Review effort level: Lite
🚦 Removed from the merge queue —
|
Conflicts in the 2.1.0 migrations, both dialects: main appended three audit_log_event index blocks while this branch appended the automations_workflow.updatedAt index. On MySQL both sides use the same `SET @ddl`/PREPARE/EXECUTE boilerplate, so the shared lines were treated as common context and the two blocks were interleaved into invalid SQL. Resolved by taking main's file verbatim and re-appending this branch's index last, so main's content stays byte-identical. Verified by running the resolved index section against live MySQL and Postgres: all four indexes create, and a second run is a no-op in both dialects. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
🔄 Playwright impact map auto-refreshedThis PR touched specs or UI source that changed the source→spec routing map. I regenerated What is this file? It is the auto-generated half of Playwright's PR planner. It routes "if source X changes, run specs Y" by walking spec imports and cross-referencing What if I want to regenerate locally instead? Run this before pushing your next change to skip the bot commit: python3 .github/scripts/generate_playwright_impact_map.py
git add .github/playwright/impact-map.generated.json
git commit --amend --no-edit # or a separate commit |
🔄 Playwright impact map auto-refreshedThis PR touched specs or UI source that changed the source→spec routing map. I regenerated What is this file? It is the auto-generated half of Playwright's PR planner. It routes "if source X changes, run specs Y" by walking spec imports and cross-referencing What if I want to regenerate locally instead? Run this before pushing your next change to skip the bot commit: python3 .github/scripts/generate_playwright_impact_map.py
git add .github/playwright/impact-map.generated.json
git commit --amend --no-edit # or a separate commit |
Code Review ✅ Approved 3 resolved / 3 findingsAdds automation Workflow retention to the Data Retention app with a configurable ✅ 3 resolved✅ Edge Case: Per-row delete failures end the workflow drain prematurely
✅ Bug: One undeletable workflow marks every retention run FAILED
✅ Quality: Poison workflow row inflates failedRecords each batch
OptionsDisplay: compact → Counting what did not apply, without listing it. Comment with these commands to change the behavior for this request:
Was this helpful? React with 👍 / 👎 | Powered by Gitar — free for open source |
|
|
Describe your changes:
Fixes #20341
automations_workflowgrows without bound. Reverse ingestion, test connection and query runnerruns each leave a
Workflowentity behind and nothing ever removes them — reverse metadata justmade it visible by creating them in volume.
Adds a
workflowRetentionPeriodsetting (default 30 days,0to retain forever) that theweekly Data Retention job drains through the app's existing batching, stats and failure-reporting
path.
Type of change:
High-level design:
Follows the built-in cleanup pattern already in
DataRetentionrather than theDataRetentionExtensionSPI — that SPI is for tables shipping outside OpenMetadata, andautomations_workflowis ours.dataRetentionConfiguration.jsonworkflowRetentionPeriod, deliberately notrequiredsrc/utils/ApplicationSchemas/DataRetentionApplication.jsonsrc/jsons/copyWorkflowDocStoreDAOs.WorkflowDAOlistIdsBeforeCutoff— one@SqlQuery, unquoted identifiers so MySQL and Postgres share itnative/2.1.0/{mysql,postgres}/schemaChanges.sqlupdatedAtfor the batch queryDataRetentioncleanAutomationWorkflows→drainInBatches→BatchDrain, underisRetentionEnabledDecisions worth reviewing:
Not
required, so no config backfill migration. jsonschema2pojo emitsprivate Integer workflowRetentionPeriod = 30;and Jackson leaves absent keys alone, so a configsaved before this change reads 30. (The only migration here is the index below.) Pinned by
workflowRetentionFallsBackToDefaultForConfigsSavedWithoutIt, and independently verified inreview against the generated POJO.
Repository delete, not bulk SQL. A
Workflowholds the service connection it ran against;dropping the row alone would strand its secrets in an external secrets manager
(
WorkflowRepository.postDelete→deleteSecretsFromWorkflow) and orphan its owner rows.Deletes are non-recursive — a Workflow has no children, and
recursiveis what makesEntityRepository.deletetake a per-row deletion lock.Selected by age alone, no status filter. Filtering to terminal statuses leaks every workflow
stuck in
Pending/Running— the same unbounded-growth bug one enum value over. A 30-day-staleupdatedAtalready means the run is dead, since any status transition bumps it.Drains on zero progress, not on a short batch. A batch can come back full and still delete
fewer rows than it fetched, because a workflow that fails to delete is skipped rather than
rethrown. Stopping on a short batch would end the run at the first failure — and since batches
are ordered oldest-first, the same undeletable row would head every batch of every run. Stopping
on zero progress continues past a partial failure and still halts on an all-poison batch instead
of spinning to the iteration cap.
A failed row is skipped, not escalated. Per-item failures go to the run's
failedRecordsonly, matching
cleanOrphanTestCasesand the other entity cleanups; the run is escalated toACTIVE_ERRORonly when a batch had rows to attempt and deleted none of them. Flipping statusper row would have marked every future run FAILED, since the same undeletable row heads every
batch. Failed ids are tracked for the run so such a row is counted once and attempted once
rather than once per batch.
Capped at 100k deletions per run. Each delete costs a repository round-trip plus a
deleteSecretsFromWorkflowcall, which is a network call on AWS/Azure setups. A large first-runbacklog spreads over several weekly runs rather than running for hours. The number is a starting
point — see the review thread.
Backward compatibility. Additive schema field, no config backfill, no API change. The index
migration is idempotent in both dialects.
Not in scope. The issue notes "once this is captured we can remove the scheduled job for
Reverse Metadata." That job isn't in this repo — there's no
REVERSE_INGESTIONproducer in OSS, itlives on the Collate side. This change is the prerequisite; removing that job is a separate
follow-up there.
Tests:
Use cases covered
workflowRetentionPeriod: 0retains automation Workflows forever.null(the app reads it as anint).Unit tests
openmetadata-service/src/test/java/.../dataRetention/DataRetentionTest.javazeroOrMissingRetentionMeansForever(pre-existing) — theisRetentionEnabledguardworkflowRetentionFallsBackToDefaultForConfigsSavedWithoutIt— the no-backfill claimaPoisonRowDoesNotStopTheWorkflowDrain— models a 50-row backlog with one undeletable row: 9 rows drained under the old short-batch predicate, 49 under the new zero-progress oneCoverage, stated honestly:
DataRetention.javais at 0.6% line coverage (2/336) from unittests — measured with
mvn -P static-code-analysis test jacoco:report, not estimated. It was 0.6%before this PR too. The class needs a live DB, the repository layer and a Quartz context, so
unit-testing the delete path would mean static-mocking
Entityand stubbing the DAO — themock-wiring test
CLAUDE.mdwarns against. The unit test above models the drain's contractrather than running
deleteExpiredWorkflows; the delete path itself is covered by the IT.This class is under the 90% target and this PR does not move it.
Backend integration tests
openmetadata-integration-tests/src/test/java/.../DataRetentionAppIT.javatest_retentionRun_cleansOldAutomationWorkflows— creates two TEST_CONNECTION workflows, backdates one 90 days inautomations_workflow, triggers the app, asserts the old row is gone and the recent one survives.Ingestion integration tests
Playwright (UI) tests
ApplicationsClassBasefrom the app schema.Manual testing performed
mvn -pl openmetadata-service compileand-pl openmetadata-integration-tests test-compile→ BUILD SUCCESS;spotless:checkclean on both.private Integer workflowRetentionPeriod = 30;— the no-backfill mechanism, confirmed in generated output rather than assumed.updatedAt→updatedat): MySQL OK, Postgres OK.SELECT 1branch /already exists, skipping), so idempotent. Dev databases restored afterwards.Limit -> Gather Merge -> Sort -> Parallel Seq ScanbecomesLimit -> Index Scan using idx_automations_workflow_updated_at.source − $id(whatparseSchemas.jsproduces), trailing newline included.UI screen recording / screenshots:
Not captured. No handwritten UI code changed — the schema adds one number input plus its help text
to the existing Data Retention app config form, rendered automatically. Happy to attach one if a
reviewer wants it.
Checklist:
Fixes <issue-number>: <short explanation>Fixes #<issue-number>above.updatedAtindex migration for both dialects; no config backfill needed — see decision 1.🤖 Generated with Claude Code