FIXES #20341 : Add reverse ingestion Workflow cleanup to Data Retention app

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/dataRetention/DataRetention.java

@@ -50,6 +50,7 @@ public class DataRetention extends AbstractNativeApplication {
   private final EntityTimeSeriesDAO testCaseResultsDAO;
   private final EntityTimeSeriesDAO profileDataDAO;
   private final CollectionDAO.AuditLogDAO auditLogDAO;
+  private final CollectionDAO.WorkflowDAO workflowDAO;
 
   public DataRetention(CollectionDAO collectionDAO, SearchRepository searchRepository) {
     super(collectionDAO, searchRepository);
@@ -59,6 +60,7 @@ public DataRetention(CollectionDAO collectionDAO, SearchRepository searchReposit
     this.testCaseResultsDAO = collectionDAO.testCaseResultTimeSeriesDao();
     this.profileDataDAO = collectionDAO.profilerDataTimeSeriesDao();
     this.auditLogDAO = collectionDAO.auditLogDAO();
+    this.workflowDAO = collectionDAO.workflowDAO();
   }
 
   @Override
@@ -122,6 +124,7 @@ private void initializeStatsDefaults() {
     entityStats.withAdditionalProperty("broken_search_entities", new StepStats());
     entityStats.withAdditionalProperty("orphaned_tag_usages", new StepStats());
     entityStats.withAdditionalProperty("audit_logs", new StepStats());
+    entityStats.withAdditionalProperty("reverse_ingestion_workflows", new StepStats());
 
     retentionStats.setEntityStats(entityStats);
   }
@@ -166,6 +169,13 @@ public void executeCleanup(DataRetentionConfiguration config) {
     LOG.info(
         "Starting cleanup for audit logs with retention period: {} days.", auditLogRetentionPeriod);
     cleanAuditLogs(auditLogRetentionPeriod);
+
+    int reverseIngestionWorkflowRetentionPeriod =
+        config.getReverseIngestionWorkflowRetentionPeriod();
+    LOG.info(
+        "Starting cleanup for reverse ingestion workflows with retention period: {} days.",
+        reverseIngestionWorkflowRetentionPeriod);
+    cleanReverseIngestionWorkflows(reverseIngestionWorkflowRetentionPeriod);
   }
 
   @Transaction
@@ -308,6 +318,19 @@ private void cleanAuditLogs(int retentionPeriod) {
     LOG.info("Audit logs cleanup complete.");
   }
 
+  @Transaction
+  private void cleanReverseIngestionWorkflows(int retentionPeriod) {
+    LOG.info(
+        "Initiating reverse ingestion workflows cleanup: Retention = {} days.", retentionPeriod);
+    long cutoffMillis = getRetentionCutoffMillis(retentionPeriod);
+
+    executeWithStatsTracking(
+        "reverse_ingestion_workflows",
+        () -> workflowDAO.deleteReverseIngestionWorkflowsBeforeCutoff(cutoffMillis, BATCH_SIZE));
+
+    LOG.info("Reverse ingestion workflows cleanup complete.");
+  }
+
   private void executeWithStatsTracking(String entity, Supplier<Integer> deleteFunction) {
     int totalDeleted = 0;
     int totalFailed = 0;

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/CollectionDAO.java

@@ -8954,6 +8954,33 @@ default Class<Workflow> getEntityClass() {
       return Workflow.class;
     }
 
+    @ConnectionAwareSqlUpdate(
+        value =
+            "DELETE FROM automations_workflow "
+                + "WHERE id IN ( "
+                + "  SELECT id FROM ( "
+                + "    SELECT id FROM automations_workflow "
+                + "    WHERE workflowType = 'REVERSE_INGESTION' "
+                + "    AND status IN ('Successful', 'Failed') "
+                + "    AND updatedAt < :cutoffTs "
+                + "    ORDER BY updatedAt LIMIT :limit "
+                + "  ) AS sub "
+                + ")",
+        connectionType = MYSQL)
+    @ConnectionAwareSqlUpdate(
+        value =
+            "DELETE FROM automations_workflow "
+                + "WHERE ctid IN ( "
+                + "  SELECT ctid FROM automations_workflow "
+                + "  WHERE workflowtype = 'REVERSE_INGESTION' "
+                + "  AND status IN ('Successful', 'Failed') "
+                + "  AND updatedat < :cutoffTs "
+                + "  ORDER BY updatedat LIMIT :limit "
+                + ")",
+        connectionType = POSTGRES)
+    int deleteReverseIngestionWorkflowsBeforeCutoff(
+        @Bind("cutoffTs") long cutoffTs, @Bind("limit") int limit);
+
     @Override
     default List<String> listBefore(
         ListFilter filter, int limit, String beforeName, String beforeId) {

openmetadata-service/src/main/resources/json/data/app/DataRetentionApplication.json

@@ -5,7 +5,9 @@
     "changeEventRetentionPeriod": 7,
     "activityThreadsRetentionPeriod": 60,
     "profileDataRetentionPeriod": 1440,
-    "testCaseResultsRetentionPeriod": 1440
+    "testCaseResultsRetentionPeriod": 1440,
+    "auditLogRetentionPeriod": 90,
+    "reverseIngestionWorkflowRetentionPeriod": 30
   },
   "appSchedule": {
     "scheduleTimeline": "Custom",

openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/dataRetention/DataRetentionAppTest.java

@@ -0,0 +1 @@
+

openmetadata-spec/src/main/resources/json/schema/entity/applications/configuration/internal/dataRetentionConfiguration.json

@@ -36,10 +36,22 @@
       "type": "integer",
       "default": 90,
       "minimum": 1
+    },
+    "reverseIngestionWorkflowRetentionPeriod": {
+      "title": "Reverse Ingestion Workflow Retention Period (days)",
+      "description": "Enter the retention period for Reverse Ingestion workflows in days (e.g., 7 for one week, 30 for one month).",
+      "type": "integer",
+      "default": 30,
+      "minimum": 1
     }
   },
   "required": [
-    "changeEventRetentionPeriod", "activityThreadsRetentionPeriod", "testCaseResultsRetentionPeriod", "profileDataRetentionPeriod", "auditLogRetentionPeriod"
+    "changeEventRetentionPeriod",
+    "activityThreadsRetentionPeriod",
+    "testCaseResultsRetentionPeriod",
+    "profileDataRetentionPeriod",
+    "auditLogRetentionPeriod",
+    "reverseIngestionWorkflowRetentionPeriod"
   ],
   "additionalProperties": false
 }

openmetadata-ui/src/main/resources/ui/public/locales/en-US/Applications/DataRetentionApplication.md

@@ -28,4 +28,18 @@ $$section
 
 Enter the retention period for Profile Data in days (e.g., 30 for one month, 60 for two months).
 
+$$
+
+$$section
+### Audit Log Retention Period (days) $(id="auditLogRetentionPeriod")
+
+Enter the retention period for Audit Log entries in days (e.g., 90 for three months).
+
+$$
+
+$$section
+### Reverse Ingestion Workflow Retention Period (days) $(id="reverseIngestionWorkflowRetentionPeriod")
+
+Enter the retention period for Reverse Ingestion workflows in days (e.g., 7 for one week, 30 for one month).
+
 $$

openmetadata-ui/src/main/resources/ui/src/generated/entity/applications/configuration/internal/dataRetentionConfiguration.ts

@@ -35,4 +35,13 @@ export interface DataRetentionConfigurationClass {
      * two months).
      */
     testCaseResultsRetentionPeriod: number;
+    /**
+     * Enter the retention period for Audit Log entries in days (e.g., 90 for three months).
+     */
+    auditLogRetentionPeriod: number;
+    /**
+     * Enter the retention period for Reverse Ingestion workflows in days (e.g., 7 for one week, 30
+     * for one month).
+     */
+    reverseIngestionWorkflowRetentionPeriod: number;
 }

openmetadata-ui/src/main/resources/ui/src/utils/ApplicationSchemas/DataRetentionApplication.json

@@ -30,13 +30,29 @@
       "description": "Enter the retention period for Profile Data in days (e.g., 30 for one month, 60 for two months).",
       "type": "integer",
       "default": 1440
+    },
+    "auditLogRetentionPeriod": {
+      "title": "Audit Log Retention Period (days)",
+      "description": "Enter the retention period for Audit Log entries in days (e.g., 90 for three months).",
+      "type": "integer",
+      "default": 90,
+      "minimum": 1
+    },
+    "reverseIngestionWorkflowRetentionPeriod": {
+      "title": "Reverse Ingestion Workflow Retention Period (days)",
+      "description": "Enter the retention period for Reverse Ingestion workflows in days (e.g., 7 for one week, 30 for one month).",
+      "type": "integer",
+      "default": 30,
+      "minimum": 1
     }
   },
   "required": [
     "changeEventRetentionPeriod",
     "activityThreadsRetentionPeriod",
     "testCaseResultsRetentionPeriod",
-    "profileDataRetentionPeriod"
+    "profileDataRetentionPeriod",
+    "auditLogRetentionPeriod",
+    "reverseIngestionWorkflowRetentionPeriod"
   ],
   "additionalProperties": false
 }