Edit TooSoonToRerunSameFlowException exception handling

Author: vsinghal85

gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/TooSoonToRerunSameFlowException.java

@@ -27,8 +27,22 @@ public class TooSoonToRerunSameFlowException extends RuntimeException {
   @Getter
   private final FlowSpec flowSpec;
 
+  /**
+   * Account for unwrapping within @{link FlowCatalog#updateOrAddSpecHelper}`s `CallbackResult` error handling for `SpecCatalogListener`s
+   * @return `TooSoonToRerunSameFlowException` wrapped in another `TooSoonToRerunSameFlowException
+   */
+  public static TooSoonToRerunSameFlowException wrappedOnce(FlowSpec flowSpec) {
+    return new TooSoonToRerunSameFlowException(flowSpec, new TooSoonToRerunSameFlowException(flowSpec));
+  }
+
   public TooSoonToRerunSameFlowException(FlowSpec flowSpec) {
     super("Lease already occupied by another recent execution of this flow: " + flowSpec);
     this.flowSpec = flowSpec;
   }
+
+  /** restricted-access ctor: use {@link #wrappedOnce(FlowSpec)} instead */
+  private TooSoonToRerunSameFlowException(FlowSpec flowSpec, Throwable cause) {
+    super("Lease already occupied by another recent execution of this flow: " + flowSpec, cause);
+    this.flowSpec = flowSpec;
+  }
 }

gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/FlowCatalog.java

@@ -57,7 +57,6 @@
 import org.apache.gobblin.runtime.api.SpecSearchObject;
 import org.apache.gobblin.runtime.api.SpecSerDe;
 import org.apache.gobblin.runtime.api.SpecStore;
-import org.apache.gobblin.runtime.api.TooSoonToRerunSameFlowException;
 import org.apache.gobblin.runtime.spec_serde.JavaSpecSerDe;
 import org.apache.gobblin.runtime.spec_store.FSSpecStore;
 import org.apache.gobblin.service.ServiceConfigKeys;
@@ -393,12 +392,7 @@ private Map<String, AddSpecResponse> updateOrAddSpecHelper(Spec spec, boolean tr
       // If flow fails compilation, the result will have a non-empty string with the error
       if (!response.getValue().getFailures().isEmpty()) {
         for (Map.Entry<SpecCatalogListener, CallbackResult<AddSpecResponse>> entry : response.getValue().getFailures().entrySet()) {
-          Throwable error = entry.getValue().getError();
-          if (error instanceof RuntimeException && error.getCause() instanceof TooSoonToRerunSameFlowException) {
-            throw (TooSoonToRerunSameFlowException) error.getCause();
-          } else {
-            throw error.getCause();
-          }
+          throw entry.getValue().getError().getCause();
         }
       }
     }

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlMultiActiveLeaseArbiter.java

@@ -362,10 +362,6 @@ else if (leaseValidityStatus == 2) {
     }
   }
 
-  /*
-    Determines if a lease can be acquired for the given flow. A lease is acquirable if
-    no existing lease record exists in arbiter table or the record is older then epsilon time
-   */
   @Override
   public boolean existsSimilarLeaseWithinConsolidationPeriod(DagActionStore.LeaseParams leaseParams) throws IOException {
     Optional<GetEventInfoResult> infoResult = getExistingEventInfo(leaseParams);

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java

@@ -24,7 +24,6 @@
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.gobblin.runtime.api.TooSoonToRerunSameFlowException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -54,6 +53,7 @@
 import org.apache.gobblin.runtime.api.SpecCatalogListener;
 import org.apache.gobblin.runtime.api.SpecProducer;
 import org.apache.gobblin.runtime.api.TopologySpec;
+import org.apache.gobblin.runtime.api.TooSoonToRerunSameFlowException;
 import org.apache.gobblin.runtime.spec_catalog.AddSpecResponse;
 import org.apache.gobblin.runtime.spec_catalog.TopologyCatalog;
 import org.apache.gobblin.service.modules.flow.FlowUtils;
@@ -139,7 +139,7 @@ public AddSpecResponse onAddSpec(Spec addedSpec) {
 
   /*
     enforces that a similar adhoc flow is not launching,
-    else throw TooSoonToRerunSameFlowException
+    else throw {@link TooSoonToRerunSameFlowException}
    */
   private void enforceNoRecentAdhocExecOfSameFlow(FlowSpec flowSpec) {
     if (!flowSpec.isScheduled()) {
@@ -151,7 +151,7 @@ private void enforceNoRecentAdhocExecOfSameFlow(FlowSpec flowSpec) {
       try {
         if (dagManagementStateStore.existsCurrentlyLaunchingExecOfSameFlow(flowGroup, flowName, FlowUtils.getOrCreateFlowExecutionId(flowSpec))) {
           _log.warn("Another recent adhoc flow execution found for " + flowGroup + "." + flowName);
-          throw new RuntimeException(new TooSoonToRerunSameFlowException(flowSpec));
+          throw TooSoonToRerunSameFlowException.wrappedOnce(flowSpec);
         }
       } catch (IOException exception) {
         _log.error("Unable to check whether similar flow exists " +  flowGroup + "." + flowName);