Updating Review Comments

Author: Suryakant Agarwal

gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/ServiceMetricNames.java

@@ -97,8 +97,8 @@ public class ServiceMetricNames {
   public static final String DAG_ACTIONS_ACT_FAILED = DAG_PROCESSING_ENGINE_PREFIX + "dagActionsActFailed.";
   public static final String DAG_ACTIONS_ACT_SUCCEEDED = DAG_PROCESSING_ENGINE_PREFIX + "dagActionsActSucceeded.";
   public static final String DAG_ACTIONS_CONCLUDE_FAILED = DAG_PROCESSING_ENGINE_PREFIX + "dagActionsConcludeFailed.";
-  public static final String DAG_ACTIONS_CONCLUDE_FLOW_SPEC_REMOVAL_SUCCEEDED = DAG_PROCESSING_ENGINE_PREFIX + "dagActionsConcludeFlowSpecRemovalSucceeded";
-  public static final String DAG_ACTIONS_CONCLUDE_FLOW_SPEC_REMOVAL_FAILED = DAG_PROCESSING_ENGINE_PREFIX + "dagActionsConcludeFlowSpecRemovalFailed";
+  public static final String DAG_ACTIONS_CONCLUDE_FLOW_SPEC_REMOVAL_SUCCEEDED = DAG_PROCESSING_ENGINE_PREFIX + "dagActionsConcludeFlowSpecRemovalSucceeded.";
+  public static final String DAG_ACTIONS_CONCLUDE_FLOW_SPEC_REMOVAL_FAILED = DAG_PROCESSING_ENGINE_PREFIX + "dagActionsConcludeFlowSpecRemovalFailed.";
   public static final String DAG_ACTIONS_CONCLUDE_SUCCEEDED = DAG_PROCESSING_ENGINE_PREFIX + "dagActionsConcludeSucceeded.";
   public static final String DAG_ACTIONS_DELETE_SUCCEEDED = DAG_PROCESSING_ENGINE_PREFIX + "dagActionsDeleteSucceeded.";
   public static final String DAG_ACTIONS_DELETE_FAILED = DAG_PROCESSING_ENGINE_PREFIX + "dagActionsDeleteFailed.";

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/modules/restli/FlowConfigsV2ResourceHandler.java

@@ -102,7 +102,7 @@ public FlowConfigsV2ResourceHandler(@Named(InjectionNames.SERVICE_NAME) String s
     this.runImmediatelyFlow = metricContext.contextAwareMeter(
         MetricRegistry.name(ServiceMetricNames.GOBBLIN_SERVICE_PREFIX, ServiceMetricNames.RUN_IMMEDIATELY_FLOW_METER));
     this.flowSpecExistsForAdhocFlow = metricContext.contextAwareMeter(
-        MetricRegistry.name(ServiceMetricNames.GOBBLIN_SERVICE_PREFIX, ServiceMetricNames. RUN_IMMEDIATELY_FLOW_METER));
+        MetricRegistry.name(ServiceMetricNames.GOBBLIN_SERVICE_PREFIX, ServiceMetricNames.FLOW_SPEC_EXISTS_FOR_ADHOC_FLOW));
   }
 
   public FlowConfig getFlowConfig(FlowId flowId)
@@ -254,12 +254,12 @@ public CreateKVResponse<ComplexResourceKey<FlowId, FlowStatusId>, FlowConfig> cr
       try {
         FlowSpec storedFlowSpec = this.flowCatalog.getSpecs(flowSpec.getUri());
         if (!storedFlowSpec.isScheduled()) {
-          log.error("FlowSpec Already Exists As Adhoc Flow  with URI: " + flowSpec.getUri());
+          log.warn("FlowSpec Already Exists As Adhoc Flow  with URI: " + flowSpec.getUri());
           if (!flowSpec.isScheduled()) {
             flowSpecExistsForAdhocFlow.mark();
           }
         } else {
-          log.error("FlowSpec Already Exists As Scheduled Flow with URI: " + flowSpec.getUri());
+          log.warn("FlowSpec Already Exists As Scheduled Flow with URI: " + flowSpec.getUri());
         }
       } catch (SpecNotFoundException e) {
         log.error("Error Retrieving FLow For Existing Flow With URI: " + flowSpec.getUri());

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

@@ -59,7 +59,8 @@ public class DagProcessingEngineMetrics {
   private final HashMap<DagActionStore.DagActionType, ContextAwareMeter> dagActionsActSucceededMeterByDagActionType =  new HashMap<>();
   private final HashMap<DagActionStore.DagActionType, ContextAwareMeter> dagActionsConcludeFailedMeterByDagActionType =  new HashMap<>();
   private final HashMap<DagActionStore.DagActionType, ContextAwareMeter> dagActionsConcludeSucceededMeterByDagActionType =  new HashMap<>();
-  private final HashMap<DagActionStore.DagActionType, ContextAwareMeter> dagActionsConcludeFlowSpecRemovalSucceededMetreByDagActionType = new HashMap<>();
+  private final HashMap<DagActionStore.DagActionType, ContextAwareMeter>
+      dagActionsConcludeFlowSpecRemovalSucceededMeterByDagActionType = new HashMap<>();
   private final HashMap<DagActionStore.DagActionType, ContextAwareMeter> dagActionsConcludeFlowSpecRemovalFailedMetreByDagActionType = new HashMap<>();
   private final HashMap<DagActionStore.DagActionType, ContextAwareMeter> dagActionsDeleteFailedMeterByDagActionType =  new HashMap<>();
   private final HashMap<DagActionStore.DagActionType, ContextAwareMeter> dagActionsDeleteSucceededMeterByDagActionType =  new HashMap<>();
@@ -91,7 +92,7 @@ public void registerAllMetrics() {
     registerMetricForEachDagActionType(this.dagActionsActSucceededMeterByDagActionType, ServiceMetricNames.DAG_ACTIONS_ACT_SUCCEEDED);
     registerMetricForEachDagActionType(this.dagActionsConcludeFailedMeterByDagActionType, ServiceMetricNames.DAG_ACTIONS_CONCLUDE_FAILED);
     registerMetricForEachDagActionType(this.dagActionsConcludeSucceededMeterByDagActionType, ServiceMetricNames.DAG_ACTIONS_CONCLUDE_SUCCEEDED);
-    registerMetricForEachDagActionType(this.dagActionsConcludeFlowSpecRemovalSucceededMetreByDagActionType, ServiceMetricNames.DAG_ACTIONS_CONCLUDE_FLOW_SPEC_REMOVAL_SUCCEEDED);
+    registerMetricForEachDagActionType(this.dagActionsConcludeFlowSpecRemovalSucceededMeterByDagActionType, ServiceMetricNames.DAG_ACTIONS_CONCLUDE_FLOW_SPEC_REMOVAL_SUCCEEDED);
     registerMetricForEachDagActionType(this.dagActionsConcludeFlowSpecRemovalFailedMetreByDagActionType, ServiceMetricNames.DAG_ACTIONS_CONCLUDE_FLOW_SPEC_REMOVAL_FAILED);
     registerMetricForEachDagActionType(this.dagActionsDeleteFailedMeterByDagActionType, ServiceMetricNames.DAG_ACTIONS_DELETE_FAILED);
     registerMetricForEachDagActionType(this.dagActionsDeleteSucceededMeterByDagActionType, ServiceMetricNames.DAG_ACTIONS_DELETE_SUCCEEDED);
@@ -169,7 +170,7 @@ public void markDagActionsConclude(DagActionStore.DagActionType dagActionType, b
 
   public void markDagActionsConcludeFlowSpecRemoval(DagActionStore.DagActionType dagActionType, boolean succeeded) {
     if (succeeded) {
-      updateMetricForDagActionType(this.dagActionsConcludeFlowSpecRemovalSucceededMetreByDagActionType, dagActionType);
+      updateMetricForDagActionType(this.dagActionsConcludeFlowSpecRemovalSucceededMeterByDagActionType, dagActionType);
     } else {
       updateMetricForDagActionType(this.dagActionsConcludeFlowSpecRemovalFailedMetreByDagActionType, dagActionType);
     }

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

@@ -62,6 +62,7 @@ public final boolean conclude() {
             log.error("Failed to Remove The FlowSpec For Adhoc Flow with URI: " + flowSpec.getUri());
             return false;
           }
+          super.dagProcEngineMetrics.markDagActionsConcludeFlowSpecRemoval(this.dagAction.getDagActionType(), true);
         }
         return true;
       }