reprocess-flag, config parameters

datahub-project · Sep 2, 2024 · df1e6e8 · df1e6e8
1 parent f158490
commit df1e6e8
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 14 deletions.
diff --git a/...src/main/java/com/linkedin/datahub/upgrade/config/BackfillDataProcessInstancesConfig.java b/...src/main/java/com/linkedin/datahub/upgrade/config/BackfillDataProcessInstancesConfig.java
@@ -22,8 +22,16 @@ public NonBlockingSystemUpgrade backfillProcessInstancesHasRunEvents(
       ElasticSearchService elasticSearchService,
       RestHighLevelClient restHighLevelClient,
       @Value("${systemUpdate.processInstanceHasRunEvents.enabled}") final boolean enabled,
+      @Value("${systemUpdate.processInstanceHasRunEvents.reprocess.enabled}")
+          boolean reprocessEnabled,
       @Value("${systemUpdate.processInstanceHasRunEvents.batchSize}") final Integer batchSize) {
     return new BackfillDataProcessInstances(
-        opContext, entityService, elasticSearchService, restHighLevelClient, enabled, batchSize);
+        opContext,
+        entityService,
+        elasticSearchService,
+        restHighLevelClient,
+        enabled,
+        reprocessEnabled,
+        batchSize);
   }
 }
diff --git a/...om/linkedin/datahub/upgrade/system/dataprocessinstances/BackfillDataProcessInstances.java b/...om/linkedin/datahub/upgrade/system/dataprocessinstances/BackfillDataProcessInstances.java
@@ -19,12 +19,18 @@ public BackfillDataProcessInstances(
       ElasticSearchService elasticSearchService,
       RestHighLevelClient restHighLevelClient,
       boolean enabled,
+      boolean reprocessEnabled,
       Integer batchSize) {
     if (enabled) {
       _steps =
           ImmutableList.of(
               new BackfillDataProcessInstancesHasRunEventsStep(
-                  opContext, entityService, elasticSearchService, restHighLevelClient, batchSize));
+                  opContext,
+                  entityService,
+                  elasticSearchService,
+                  restHighLevelClient,
+                  reprocessEnabled,
+                  batchSize));
     } else {
       _steps = ImmutableList.of();
     }

diff --git a/...hub/upgrade/system/dataprocessinstances/BackfillDataProcessInstancesHasRunEventsStep.java b/...hub/upgrade/system/dataprocessinstances/BackfillDataProcessInstancesHasRunEventsStep.java
@@ -1,7 +1,6 @@
 package com.linkedin.datahub.upgrade.system.dataprocessinstances;
 
-import static com.linkedin.metadata.Constants.DATA_PROCESS_INSTANCE_ENTITY_NAME;
-import static com.linkedin.metadata.Constants.DATA_PROCESS_INSTANCE_RUN_EVENT_ASPECT_NAME;
+import static com.linkedin.metadata.Constants.*;
 
 import com.google.common.base.Throwables;
 import com.linkedin.common.AuditStamp;
@@ -49,18 +48,21 @@ public class BackfillDataProcessInstancesHasRunEventsStep implements UpgradeStep
   private final ElasticSearchService elasticSearchService;
   private final RestHighLevelClient restHighLevelClient;
 
+  private final boolean reprocessEnabled;
   private final Integer batchSize;
 
   public BackfillDataProcessInstancesHasRunEventsStep(
       OperationContext opContext,
       EntityService<?> entityService,
       ElasticSearchService elasticSearchService,
       RestHighLevelClient restHighLevelClient,
+      boolean reprocessEnabled,
       Integer batchSize) {
     this.opContext = opContext;
     this.entityService = entityService;
     this.elasticSearchService = elasticSearchService;
     this.restHighLevelClient = restHighLevelClient;
+    this.reprocessEnabled = reprocessEnabled;
     this.batchSize = batchSize;
   }
 
@@ -156,12 +158,7 @@ public boolean isOptional() {
   /** Returns whether the upgrade should be skipped. */
   @Override
   public boolean skip(UpgradeContext context) {
-    return false;
-
-    /*boolean envEnabled =
-        Boolean.parseBoolean(System.getenv("BACKFILL_PROCESS_INSTANCES_HAS_RUN_EVENTS"));
-
-    if (envEnabled) {
+    if (reprocessEnabled) {
       return false;
     }
 
@@ -170,8 +167,7 @@ public boolean skip(UpgradeContext context) {
             context.opContext(), UPGRADE_ID_URN, DATA_HUB_UPGRADE_RESULT_ASPECT_NAME, true);
     if (previouslyRun) {
       log.info("{} was already run. Skipping.", id());
-      return true;
     }
-    return false;*/
+    return previouslyRun;
   }
 }
diff --git a/metadata-service/configuration/src/main/resources/application.yaml b/metadata-service/configuration/src/main/resources/application.yaml
@@ -351,8 +351,10 @@ systemUpdate:
     reprocess:
       enabled: ${BOOTSTRAP_SYSTEM_UPDATE_OWNERSHIP_TYPES_REPROCESS:false}
   processInstanceHasRunEvents:
-    enabled: true
-    batchSize: 50
+    enabled: ${BOOTSTRAP_SYSTEM_UPDATE_PROCESS_INSTANCE_HAS_RUN_EVENTS_ENABLED:true}
+    batchSize: ${BOOTSTRAP_SYSTEM_UPDATE_PROCESS_INSTANCE_HAS_RUN_EVENTS_BATCH_SIZE:100}
+    reprocess:
+      enabled: ${BOOTSTRAP_SYSTEM_UPDATE_PROCESS_INSTANCE_HAS_RUN_EVENTS_REPROCESS:false}
 
 structuredProperties:
   enabled: ${ENABLE_STRUCTURED_PROPERTIES_HOOK:true} # applies structured properties mappings