Extract TooSoonToRerunSameFlowException from Throwable in Flow Catalog listeners

Author: vsinghal85

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

@@ -29,6 +29,7 @@
 import java.util.Properties;
 
 import org.apache.commons.lang3.reflect.ConstructorUtils;
+import org.apache.gobblin.runtime.api.TooSoonToRerunSameFlowException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -392,7 +393,12 @@ 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()) {
-          throw entry.getValue().getError().getCause();
+          Throwable error = entry.getValue().getError();
+          if (error instanceof TooSoonToRerunSameFlowException) {
+            throw (TooSoonToRerunSameFlowException) error;
+          } else {
+            throw error.getCause();
+          }
         }
       }
     }

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

@@ -41,7 +41,6 @@
 import org.apache.gobblin.runtime.api.SpecNotFoundException;
 import org.apache.gobblin.runtime.api.TopologySpec;
 import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
-import org.apache.gobblin.service.modules.flow.FlowUtils;
 import org.apache.gobblin.service.modules.flowgraph.Dag;
 import org.apache.gobblin.service.modules.flowgraph.DagNodeId;
 import org.apache.gobblin.service.modules.spec.JobExecutionPlan;