[GOBBLIN-2129] avoid throwing unnessary exception that is polluting l…

…ogs (apache#4023)

gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java

@@ -115,6 +115,7 @@ public class GobblinServiceJobScheduler extends JobScheduler implements SpecCata
   protected final Boolean isWarmStandbyEnabled;
   protected final Optional<UserQuotaManager> quotaManager;
   protected final Optional<FlowLaunchHandler> flowTriggerHandler;
+  // todo - consider using JobScheduler::scheduledJobs in place of scheduledFlowSpecs
   @Getter
   protected final Map<String, FlowSpec> scheduledFlowSpecs;
   @Getter
@@ -617,7 +618,7 @@ public AddSpecResponse onAddSpec(Spec addedSpec) {
    * @param specURI
    * @param specVersion
    */
-  private void unscheduleSpec(URI specURI, String specVersion) throws JobException {
+  private boolean unscheduleSpec(URI specURI, String specVersion) throws JobException {
     if (this.scheduledFlowSpecs.containsKey(specURI.toString())) {
       _log.info("Unscheduling flowSpec " + specURI + "/" + specVersion);
       this.scheduledFlowSpecs.remove(specURI.toString());
@@ -630,10 +631,12 @@ private void unscheduleSpec(URI specURI, String specVersion) throws JobException
         } catch (SpecNotFoundException e) {
           _log.warn("Unable to retrieve spec for URI {}", specURI);
         }
+      return true;
     } else {
-      throw new JobException(String.format(
-          "Spec with URI: %s was not found in cache. May be it was cleaned, if not please clean it manually",
+      _log.info(String.format(
+          "Spec with URI: %s was not found in cache. Maybe it was cleaned, if not please clean it manually",
           specURI));
+      return false;
     }
   }
 
@@ -659,8 +662,9 @@ public void onDeleteSpec(URI deletedSpecURI, String deletedSpecVersion, Properti
 
     try {
       Spec deletedSpec = this.scheduledFlowSpecs.get(deletedSpecURI.toString());
-      unscheduleSpec(deletedSpecURI, deletedSpecVersion);
-      this.orchestrator.remove(deletedSpec, headers);
+      if (unscheduleSpec(deletedSpecURI, deletedSpecVersion)) {
+        this.orchestrator.remove(deletedSpec, headers);
+      }
     } catch (JobException | IOException e) {
       _log.warn(String.format("Spec with URI: %s was not unscheduled cleaning", deletedSpecURI), e);
     }