[GOBBLIN-2144] Prevent GenerateWorkUnitsImpl from inadvertently cle…

…aning up temp/staging dirs (#4039)
apache · Aug 27, 2024 · 444f266 · 444f266
1 parent b28b468
commit 444f266
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 12 deletions.
diff --git a/...in/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImpl.java b/...in/java/org/apache/gobblin/service/modules/orchestration/DagManagementTaskStreamImpl.java
@@ -110,6 +110,7 @@ public DagManagementTaskStreamImpl(Config config, Optional<DagActionStore> dagAc
     this.dagProcEngineMetrics = dagProcEngineMetrics;
   }
 
+  @Override
   public synchronized void addDagAction(DagActionStore.LeaseParams leaseParams) {
     log.info("Adding {} to queue...", leaseParams);
     if (!this.leaseParamsQueue.offer(leaseParams)) {

diff --git a/...al/src/main/java/org/apache/gobblin/temporal/ddm/activity/impl/GenerateWorkUnitsImpl.java b/...al/src/main/java/org/apache/gobblin/temporal/ddm/activity/impl/GenerateWorkUnitsImpl.java
@@ -30,7 +30,6 @@
 import io.temporal.failure.ApplicationFailure;
 import lombok.extern.slf4j.Slf4j;
 
-import org.apache.gobblin.commit.DeliverySemantics;
 import org.apache.gobblin.converter.initializer.ConverterInitializerFactory;
 import org.apache.gobblin.destination.DestinationDatasetHandlerService;
 import org.apache.gobblin.metrics.event.EventSubmitter;
@@ -117,9 +116,9 @@ protected static List<WorkUnit> generateWorkUnitsForJobState(JobState jobState,
 
     // TODO: count total bytes for progress tracking!
 
-    boolean canCleanUp = canCleanStagingData(jobState);
+    boolean canCleanUpTempDirs = false; // unlike `AbstractJobLauncher` running the job end-to-end, this is Work Discovery only, so WAY TOO SOON for cleanup
     DestinationDatasetHandlerService datasetHandlerService = closer.register(
-        new DestinationDatasetHandlerService(jobState, canCleanUp, eventSubmitterContext.create()));
+        new DestinationDatasetHandlerService(jobState, canCleanUpTempDirs, eventSubmitterContext.create()));
     WorkUnitStream handledWorkUnitStream = datasetHandlerService.executeHandlers(workUnitStream);
 
     // initialize writer and converter(s)
@@ -143,13 +142,4 @@ protected static List<WorkUnit> generateWorkUnitsForJobState(JobState jobState,
 
     return AbstractJobLauncher.materializeWorkUnitList(trackedWorkUnitStream);
   }
-
-  protected static boolean canCleanStagingData(JobState jobState) {
-    if (DeliverySemantics.EXACTLY_ONCE.equals(DeliverySemantics.parse(jobState))) {
-      String errMsg = "DeliverySemantics.EXACTLY_ONCE NOT currently supported; job " + jobState.getJobId();
-      log.error(errMsg);
-      throw ApplicationFailure.newNonRetryableFailure(errMsg, "Unsupported: DeliverySemantics.EXACTLY_ONCE");
-    }
-    return true;
-  }
 }