[GOBBLIN-2220]: Send Metrics When Flow Spec Already Exists For An Adhoc Flow #4136

aga9900 · 2025-08-26T07:40:07Z

Dear Gobblin maintainers,

Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!

JIRA

My PR addresses the following Gobblin JIRA issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-2220

Description

Here are some details about my PR, including screenshots (if applicable):
This PR does 2 things:

Send Metrics When 409 is sent to Users In Case Of Conflicting Flows.
Send metrics When Flow Spec Deletion Fails for an Adhoc Flow.

Tests

My PR adds the following unit tests OR does not need testing for this extremely good reason:
Only Adding New Metrics and Logging.

Commits

My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"

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

...er/src/main/java/org/apache/gobblin/service/modules/restli/FlowConfigsV2ResourceHandler.java

...n/java/org/apache/gobblin/service/modules/orchestration/task/DagProcessingEngineMetrics.java

khandelwal-prateek · 2025-08-27T16:22:12Z

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

  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.";


success is the expected/default case, so the success metric doesn't give us any additional signal. The failure metric is actionable, and tracking failures should be sufficient here. I would suggest dropping success metric for this one as it just adds noise

khandelwal-prateek · 2025-08-27T16:40:49Z

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

+            dagManagementStateStore.removeFlowSpec(flowSpec.getUri(), new Properties(), false);
+          } catch (Exception e) {
+            super.dagProcEngineMetrics.markDagActionsConcludeFlowSpecRemoval(this.dagAction.getDagActionType(), false);
+            log.error("Failed to Remove The FlowSpec For Adhoc Flow with URI: " + flowSpec.getUri());


Please log the exception as well using log.error(..., e), so that stack trace is captured.

khandelwal-prateek · 2025-08-27T16:51:55Z

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

+          } catch (Exception e) {
+            super.dagProcEngineMetrics.markDagActionsConcludeFlowSpecRemoval(this.dagAction.getDagActionType(), false);
+            log.error("Failed to Remove The FlowSpec For Adhoc Flow with URI: " + flowSpec.getUri());
+            return false;


earlier the RuntimeException was not caught here, so it was getting caught in DagProcessingEngine which was marking dagManagementStateStore.getDagManagerMetrics().dagProcessingExceptionMeter.mark();, now we are gulping the exception so it would not be handled in DagProcessingEngine, we should re-throw the exception here

khandelwal-prateek · 2025-08-27T16:53:09Z

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

+            log.error("Failed to Remove The FlowSpec For Adhoc Flow with URI: " + flowSpec.getUri());
+            return false;
+          }
+          super.dagProcEngineMetrics.markDagActionsConcludeFlowSpecRemoval(this.dagAction.getDagActionType(), true);


imo, we don't need to mark for success

khandelwal-prateek · 2025-08-27T17:10:40Z

...er/src/main/java/org/apache/gobblin/service/modules/restli/FlowConfigsV2ResourceHandler.java

+      } catch (SpecNotFoundException e) {
+        log.error("Error Retrieving FLow For Existing Flow With URI: " + flowSpec.getUri());


this is not required, since the if block has already checked for existence of flowSpec

khandelwal-prateek · 2025-08-27T17:13:08Z

...er/src/main/java/org/apache/gobblin/service/modules/restli/FlowConfigsV2ResourceHandler.java

+            flowSpecExistsForAdhocFlow.mark();
+          }
+        } else {
+          log.warn("FlowSpec Already Exists As Scheduled Flow with URI: " + flowSpec.getUri());


please move log from line 253 to here, since we are logging twice now

khandelwal-prateek · 2025-08-28T03:26:10Z

...er/src/main/java/org/apache/gobblin/service/modules/restli/FlowConfigsV2ResourceHandler.java

+        FlowSpec storedFlowSpec = this.flowCatalog.getSpecs(flowSpec.getUri());
+        if (!storedFlowSpec.isScheduled()) {
+          log.warn("FlowSpec Already Exists As Adhoc Flow  with URI: " + flowSpec.getUri());
+          if (!flowSpec.isScheduled()) {


if it was an adhoc flow(!storedFlowSpec.isScheduled()) and flowSpec is not deleted, that should be sufficient condition to mark the metric for unexpected behaviour. We don't need to check if current flowSpec is scheduled or not

khandelwal-prateek · 2025-08-28T03:33:38Z

...er/src/main/java/org/apache/gobblin/service/modules/restli/FlowConfigsV2ResourceHandler.java

+        if (!storedFlowSpec.isScheduled()) {
+          log.warn("FlowSpec Already Exists As Adhoc Flow  with URI: " + flowSpec.getUri());
+          if (!flowSpec.isScheduled()) {
+            flowSpecExistsForAdhocFlow.mark();


there’s an expected case where a FlowSpec isn't deleted for an adhoc flow, specifically when a new flow is triggered before the previous one is launched. In that case, we throw a TooSoonToRerunSameFlowException from Orchestrator.onAddSpec. We should exclude these scenarios, since these are valid, and only flag cases where the FlowSpec wasn't deleted even after the DAG was launched, which is unexpected.

This would mark the metric for both the scenarios and we would have false/non-actionable alerts in such cases

Suryakant Agarwal added 9 commits August 13, 2025 12:54

Fix Bug For Executions Failing With Only One Schedule

5cdca97

Updating Gobblin Checkstyle

436e081

Update Review Comments

9940fcd

Removing Repetative ClassPaths From Test

506a2e0

Merge branch 'master' of https://github.com/apache/gobblin

5a5efdd

Add Metric When A FLow Spec Already Exists For A New Flow

b5a946d

Adding Metrics For Ahhoc Flow Scep Failure and 409 sent to Users

3d7c21d

Fixing Code Style

1e8d444

Fixing Code Style For FlowConfigsV2ResourceHandler

9d1c02e

aga9900 changed the title ~~[GOBBLIN-2220]: Send Metrics When an adhoc Flow Spec Already Exists For Adhoc Flow (edited)~~ [GOBBLIN-2220]: Send Metrics When an adhoc Flow Spec Already Exists For Adhoc Flow Aug 26, 2025

Changing Method Name and Code Style

f5e8dc3

aga9900 changed the title ~~[GOBBLIN-2220]: Send Metrics When an adhoc Flow Spec Already Exists For Adhoc Flow~~ [GOBBLIN-2220]: Send Metrics When Flow Spec Already Exists For An Adhoc Flow Aug 26, 2025

abhishekmjain reviewed Aug 26, 2025

View reviewed changes

Updating Review Comments

bb0ba83

aga9900 force-pushed the master branch from 231c16e to bb0ba83 Compare August 26, 2025 08:29

Updating Review Comments

05af792

abhishekmjain approved these changes Aug 27, 2025

View reviewed changes

khandelwal-prateek suggested changes Aug 27, 2025

View reviewed changes

khandelwal-prateek reviewed Aug 28, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[GOBBLIN-2220]: Send Metrics When Flow Spec Already Exists For An Adhoc Flow #4136

[GOBBLIN-2220]: Send Metrics When Flow Spec Already Exists For An Adhoc Flow #4136

Uh oh!

aga9900 commented Aug 26, 2025 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

khandelwal-prateek Aug 27, 2025

Uh oh!

khandelwal-prateek Aug 27, 2025

Uh oh!

khandelwal-prateek Aug 27, 2025

Uh oh!

khandelwal-prateek Aug 27, 2025

Uh oh!

khandelwal-prateek Aug 27, 2025

Uh oh!

khandelwal-prateek Aug 27, 2025

Uh oh!

khandelwal-prateek Aug 28, 2025

Uh oh!

khandelwal-prateek Aug 28, 2025

Uh oh!

Uh oh!

		} catch (SpecNotFoundException e) {
		log.error("Error Retrieving FLow For Existing Flow With URI: " + flowSpec.getUri());

[GOBBLIN-2220]: Send Metrics When Flow Spec Already Exists For An Adhoc Flow #4136

Are you sure you want to change the base?

[GOBBLIN-2220]: Send Metrics When Flow Spec Already Exists For An Adhoc Flow #4136

Uh oh!

Conversation

aga9900 commented Aug 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

JIRA

Description

Tests

Commits

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

aga9900 commented Aug 26, 2025 •

edited

Loading