You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by "umustafi (via GitHub)" <gi...@apache.org> on 2024/07/09 00:29:45 UTC
[PR] [GOBBLIN-2107] Delete adhoc flowSpecs from flowCatalog bug fix [gobblin]
umustafi opened a new pull request, #3996:
URL: https://github.com/apache/gobblin/pull/3996
* if job compilation fails then ensure quota is released and flow compilation event sent in all cases
* fix bug where adhoc flows are not deleted in multi-active scheduler case by re-adding deletion to DagManager.addDag()
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
- [X] My PR addresses the following [Gobblin JIRA](https://issues.apache.org/jira/browse/GOBBLIN/) issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-2107
### Description
- [X] Here are some details about my PR, including screenshots (if applicable):
Delete adhoc `flowSpecs` from `flowCatalog` to avoid build up of adhoc flowSpecs in catalog even when job compilation fails by adding the deletion in a finally block in non-MA scheduler case.
1. Previous PR: https://github.com/apache/gobblin/pull/3944/files causes a bug where adhoc flows are not deleted when multi-active scheduler is enabled. This PR re-adds deletion of `flowSpec` to the `dagManager` to be called by the active host for the MA scheduler case (there will only be one active host if `dagManager` is enabled otherwise code will use `dagProcessingEngine`).
2. It also ensures quota is released and failed flow compilation event sent by the `FlowCompilationValidationHelper` in all compilation failure cases. This was being incorrectly handled before.
### Tests
- [X] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
### Commits
- [X] 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](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
3. Subject is limited to 50 characters
4. Subject does not end with a period
5. Subject uses the imperative mood ("add", not "adding")
6. Body wraps at 72 characters
7. Body explains "what" and "why", not "how"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2107] Delete adhoc flowSpecs from flowCatalog bug fix [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3996:
URL: https://github.com/apache/gobblin/pull/3996#discussion_r1670943559
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/utils/FlowCompilationValidationHelper.java:
##########
@@ -140,31 +140,39 @@ public Optional<Dag<JobExecutionPlan>> validateAndHandleConcurrentExecution(Conf
ConfigurationKeys.FLOW_ALLOW_CONCURRENT_EXECUTION, String.valueOf(this.isFlowConcurrencyEnabled)));
Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.compileFlow(flowSpec);
- if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
- return Optional.absent();
- }
- addFlowExecutionIdIfAbsent(flowMetadata, jobExecutionPlanDag);
-
- if (isExecutionPermitted(flowStatusGenerator, flowGroup, flowName, allowConcurrentExecution,
- Long.parseLong(flowMetadata.get(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD)))) {
- return Optional.fromNullable(jobExecutionPlanDag);
- } else {
- log.warn("Another instance of flowGroup: {}, flowName: {} running; Skipping flow execution since "
- + "concurrent executions are disabled for this flow.", flowGroup, flowName);
- sharedFlowMetricsSingleton.conditionallyUpdateFlowGaugeSpecState(flowSpec,
- SharedFlowMetricsSingleton.CompiledState.SKIPPED);
- Instrumented.markMeter(sharedFlowMetricsSingleton.getSkippedFlowsMeter());
- if (!flowSpec.isScheduled()) {
- // For ad-hoc flow, we might already increase quota, we need to decrease here
- for (Dag.DagNode dagNode : jobExecutionPlanDag.getStartNodes()) {
- quotaManager.releaseQuota(dagNode);
+ Optional<Dag<JobExecutionPlan>> optionalJobExecutionPlan = null;
+
+ try {
+ if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
+ optionalJobExecutionPlan = Optional.absent();
Review Comment:
this logic gets convoluted. why not return the `Optional.absent()` as previously and leave it to a `finally` to clean up?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java:
##########
@@ -321,8 +321,10 @@ protected void submitFlowToDagManagerHelper(DagActionStore.DagAction dagAction,
launchSubmissionMetricProxy.markFailure();
return;
} catch (SpecNotFoundException e) {
- log.warn("Spec not found for flowId {} due to exception {}", flowId, e.getMessage());
- launchSubmissionMetricProxy.markFailure();
+ log.info("Spec not found for flowId {} due to deletion by active dagManager host due to exception {}",
+ flowId, e.getMessage());
+ // TODO: mark this failure if there are other valid cases of this exception
+ // launchSubmissionMetricProxy.markFailure();
Review Comment:
1. let's figure out whether there are other valid cases!
not sure if I'm misunderstanding... but a TODO is far more reasonable when the impl change would be considerable effort, so we plan to come back, vs. when the task is to figure out how our system behaves. usually, the latter is worth resolving ASAP.
give that, was it incorrect previously to `markFailure()` and we no longer want to do?
2. on the message: `"Spec not found for flowId {} due to deletion by active dagManager host due to exception {}",`
is that certain it was due to deletion by active DM or is that speculative?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/utils/FlowCompilationValidationHelper.java:
##########
@@ -140,31 +140,39 @@ public Optional<Dag<JobExecutionPlan>> validateAndHandleConcurrentExecution(Conf
ConfigurationKeys.FLOW_ALLOW_CONCURRENT_EXECUTION, String.valueOf(this.isFlowConcurrencyEnabled)));
Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.compileFlow(flowSpec);
- if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
- return Optional.absent();
- }
- addFlowExecutionIdIfAbsent(flowMetadata, jobExecutionPlanDag);
-
- if (isExecutionPermitted(flowStatusGenerator, flowGroup, flowName, allowConcurrentExecution,
- Long.parseLong(flowMetadata.get(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD)))) {
- return Optional.fromNullable(jobExecutionPlanDag);
- } else {
- log.warn("Another instance of flowGroup: {}, flowName: {} running; Skipping flow execution since "
- + "concurrent executions are disabled for this flow.", flowGroup, flowName);
- sharedFlowMetricsSingleton.conditionallyUpdateFlowGaugeSpecState(flowSpec,
- SharedFlowMetricsSingleton.CompiledState.SKIPPED);
- Instrumented.markMeter(sharedFlowMetricsSingleton.getSkippedFlowsMeter());
- if (!flowSpec.isScheduled()) {
- // For ad-hoc flow, we might already increase quota, we need to decrease here
- for (Dag.DagNode dagNode : jobExecutionPlanDag.getStartNodes()) {
- quotaManager.releaseQuota(dagNode);
+ Optional<Dag<JobExecutionPlan>> optionalJobExecutionPlan = null;
+
+ try {
+ if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
+ optionalJobExecutionPlan = Optional.absent();
+ }
+ addFlowExecutionIdIfAbsent(flowMetadata, jobExecutionPlanDag);
+
+ if (isExecutionPermitted(flowStatusGenerator, flowGroup, flowName, allowConcurrentExecution,
+ Long.parseLong(flowMetadata.get(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD)))) {
+ optionalJobExecutionPlan = Optional.fromNullable(jobExecutionPlanDag);
+ } else {
+ log.warn("Another instance of flowGroup: {}, flowName: {} running; Skipping flow execution since "
+ + "concurrent executions are disabled for this flow.", flowGroup, flowName);
+ optionalJobExecutionPlan = Optional.absent();
+ }
+ } finally {
+ if (optionalJobExecutionPlan == null || !optionalJobExecutionPlan.isPresent()) {
Review Comment:
again, convoluted to need to handle a case where a var may not yet have been initialized (even ironic, since the point of `Optional` is that we no longer need `null` checks)
where are the test cases to validate correctness?
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -289,7 +293,7 @@ public void submitFlowToDagManager(FlowSpec flowSpec, Dag<JobExecutionPlan> jobE
Note that the responsibility of the multi-active scheduler mode ends after this method is completed AND the
consumption of a launch type event is committed to the consumer.
*/
- this.dagManager.addDag(jobExecutionPlanDag, true, true);
+ this.dagManager.addDagAndRemoveAdhocFlowSpec(flowSpec, jobExecutionPlanDag, true, true);
Review Comment:
now the attempt to remove an ad hoc spec would occur twice, since it's also on line 267.
I believe your aim is to have removal also happen in the path from [DagActionStoreChangeMonitor](https://github.com/apache/gobblin/blob/4721d086f78553cab28e46bf406787df6cc2dea8/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java#L318), correct?
couldn't that run into the same issue, where if the flow compilation validation helper fails (see L281 directly above), then the call to this new combo addDag+removeAdHoc would be skipped?
this new PR helps me realize I had missed the `DagActionStoreChangeMonitor` case in my recent PR. the crux of that one was to divide adding the dag from removing the ad hoc spec. the latter MUST happen, suggesting it belongs inside a `finally`, whereas the former is only best-effort.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2107] Delete adhoc flowSpecs from flowCatalog bug fix [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3996:
URL: https://github.com/apache/gobblin/pull/3996#discussion_r1671460273
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -271,14 +275,25 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
- public void submitFlowToDagManager(FlowSpec flowSpec) throws IOException, InterruptedException {
- Optional<Dag<JobExecutionPlan>> optionalJobExecutionPlanDag =
- this.flowCompilationValidationHelper.createExecutionPlanIfValid(flowSpec);
- if (optionalJobExecutionPlanDag.isPresent()) {
- submitFlowToDagManager(flowSpec, optionalJobExecutionPlanDag.get());
- } else {
- _log.warn("Flow: {} submitted to dagManager failed to compile and produce a job execution plan dag", flowSpec);
- Instrumented.markMeter(this.flowOrchestrationFailedMeter);
+ /**
+ * Method that accepts a flowSpec that it compiles before forwarding to the DagManagerfor execution. It's meant to be
+ * called by {@link org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor}
+ * @param flowSpec
+ * @throws IOException
+ * @throws InterruptedException
Review Comment:
suggest to simplify to:
```
* Compiles the provided {@link FlowSpec} into a {@link Dag<JobExecutionPlan>} and forwards that to the {@link DagManager} for execution. It's meant to be
* called by {@link org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor}
```
tip: avoid "Method that... " or "Class that..." or "Code to..." in javadoc. since that could be said of *every* method/class, let's just take it as an implied given.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2107] Delete adhoc flowSpecs from flowCatalog bug fix [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3996:
URL: https://github.com/apache/gobblin/pull/3996#discussion_r1671460273
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -271,14 +275,25 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
- public void submitFlowToDagManager(FlowSpec flowSpec) throws IOException, InterruptedException {
- Optional<Dag<JobExecutionPlan>> optionalJobExecutionPlanDag =
- this.flowCompilationValidationHelper.createExecutionPlanIfValid(flowSpec);
- if (optionalJobExecutionPlanDag.isPresent()) {
- submitFlowToDagManager(flowSpec, optionalJobExecutionPlanDag.get());
- } else {
- _log.warn("Flow: {} submitted to dagManager failed to compile and produce a job execution plan dag", flowSpec);
- Instrumented.markMeter(this.flowOrchestrationFailedMeter);
+ /**
+ * Method that accepts a flowSpec that it compiles before forwarding to the DagManagerfor execution. It's meant to be
+ * called by {@link org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor}
+ * @param flowSpec
+ * @throws IOException
+ * @throws InterruptedException
Review Comment:
suggest to simplify to:
```
* Compiles the provided {@link FlowSpec} into a {@link Dag<JobExecutionPlan>} it forwards to the {@link DagManager} for execution. It's meant to be
* called by {@link org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor}
```
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -259,8 +261,10 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
// Depending on if DagManager is present, handle execution
submitFlowToDagManager(flowSpec, compiledDag);
} finally {
- // remove from the flow catalog, regardless of whether the flow was successfully validated and permitted to exec (concurrently)
- this.dagManager.removeFlowSpecIfAdhoc(flowSpec);
+ /* Remove adhoc flow spec from the flow catalog, regardless of whether the flow was successfully validated
+ and permitted to exec (concurrently)
+ */
+ this.dagManager.removeFlowSpecIfAdhoc(flowSpec);
Review Comment:
indentation looks off, since it lines up w/ the closing `}` that follows
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2107] Delete adhoc flowSpecs from flowCatalog bug fix [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3996:
URL: https://github.com/apache/gobblin/pull/3996#discussion_r1671397418
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/utils/FlowCompilationValidationHelper.java:
##########
@@ -140,31 +140,39 @@ public Optional<Dag<JobExecutionPlan>> validateAndHandleConcurrentExecution(Conf
ConfigurationKeys.FLOW_ALLOW_CONCURRENT_EXECUTION, String.valueOf(this.isFlowConcurrencyEnabled)));
Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.compileFlow(flowSpec);
- if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
- return Optional.absent();
- }
- addFlowExecutionIdIfAbsent(flowMetadata, jobExecutionPlanDag);
-
- if (isExecutionPermitted(flowStatusGenerator, flowGroup, flowName, allowConcurrentExecution,
- Long.parseLong(flowMetadata.get(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD)))) {
- return Optional.fromNullable(jobExecutionPlanDag);
- } else {
- log.warn("Another instance of flowGroup: {}, flowName: {} running; Skipping flow execution since "
- + "concurrent executions are disabled for this flow.", flowGroup, flowName);
- sharedFlowMetricsSingleton.conditionallyUpdateFlowGaugeSpecState(flowSpec,
- SharedFlowMetricsSingleton.CompiledState.SKIPPED);
- Instrumented.markMeter(sharedFlowMetricsSingleton.getSkippedFlowsMeter());
- if (!flowSpec.isScheduled()) {
- // For ad-hoc flow, we might already increase quota, we need to decrease here
- for (Dag.DagNode dagNode : jobExecutionPlanDag.getStartNodes()) {
- quotaManager.releaseQuota(dagNode);
+ Optional<Dag<JobExecutionPlan>> optionalJobExecutionPlan = null;
+
+ try {
+ if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
+ optionalJobExecutionPlan = Optional.absent();
+ }
+ addFlowExecutionIdIfAbsent(flowMetadata, jobExecutionPlanDag);
+
+ if (isExecutionPermitted(flowStatusGenerator, flowGroup, flowName, allowConcurrentExecution,
+ Long.parseLong(flowMetadata.get(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD)))) {
+ optionalJobExecutionPlan = Optional.fromNullable(jobExecutionPlanDag);
+ } else {
+ log.warn("Another instance of flowGroup: {}, flowName: {} running; Skipping flow execution since "
+ + "concurrent executions are disabled for this flow.", flowGroup, flowName);
+ optionalJobExecutionPlan = Optional.absent();
+ }
+ } finally {
+ if (optionalJobExecutionPlan == null || !optionalJobExecutionPlan.isPresent()) {
Review Comment:
updated this code path to make it more clear. I realized the quota manager does not need to be updated when the compilation fails so this looks much simpler now
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/utils/FlowCompilationValidationHelper.java:
##########
@@ -140,31 +140,39 @@ public Optional<Dag<JobExecutionPlan>> validateAndHandleConcurrentExecution(Conf
ConfigurationKeys.FLOW_ALLOW_CONCURRENT_EXECUTION, String.valueOf(this.isFlowConcurrencyEnabled)));
Dag<JobExecutionPlan> jobExecutionPlanDag = specCompiler.compileFlow(flowSpec);
- if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
- return Optional.absent();
- }
- addFlowExecutionIdIfAbsent(flowMetadata, jobExecutionPlanDag);
-
- if (isExecutionPermitted(flowStatusGenerator, flowGroup, flowName, allowConcurrentExecution,
- Long.parseLong(flowMetadata.get(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD)))) {
- return Optional.fromNullable(jobExecutionPlanDag);
- } else {
- log.warn("Another instance of flowGroup: {}, flowName: {} running; Skipping flow execution since "
- + "concurrent executions are disabled for this flow.", flowGroup, flowName);
- sharedFlowMetricsSingleton.conditionallyUpdateFlowGaugeSpecState(flowSpec,
- SharedFlowMetricsSingleton.CompiledState.SKIPPED);
- Instrumented.markMeter(sharedFlowMetricsSingleton.getSkippedFlowsMeter());
- if (!flowSpec.isScheduled()) {
- // For ad-hoc flow, we might already increase quota, we need to decrease here
- for (Dag.DagNode dagNode : jobExecutionPlanDag.getStartNodes()) {
- quotaManager.releaseQuota(dagNode);
+ Optional<Dag<JobExecutionPlan>> optionalJobExecutionPlan = null;
+
+ try {
+ if (jobExecutionPlanDag == null || jobExecutionPlanDag.isEmpty()) {
+ optionalJobExecutionPlan = Optional.absent();
+ }
+ addFlowExecutionIdIfAbsent(flowMetadata, jobExecutionPlanDag);
+
+ if (isExecutionPermitted(flowStatusGenerator, flowGroup, flowName, allowConcurrentExecution,
+ Long.parseLong(flowMetadata.get(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD)))) {
+ optionalJobExecutionPlan = Optional.fromNullable(jobExecutionPlanDag);
+ } else {
+ log.warn("Another instance of flowGroup: {}, flowName: {} running; Skipping flow execution since "
+ + "concurrent executions are disabled for this flow.", flowGroup, flowName);
+ optionalJobExecutionPlan = Optional.absent();
+ }
+ } finally {
+ if (optionalJobExecutionPlan == null || !optionalJobExecutionPlan.isPresent()) {
Review Comment:
updated this code path to make it more clear. I realized the quota manager does not need to be updated when the compilation fails so this looks much simpler now
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java:
##########
@@ -321,8 +321,10 @@ protected void submitFlowToDagManagerHelper(DagActionStore.DagAction dagAction,
launchSubmissionMetricProxy.markFailure();
return;
} catch (SpecNotFoundException e) {
- log.warn("Spec not found for flowId {} due to exception {}", flowId, e.getMessage());
- launchSubmissionMetricProxy.markFailure();
+ log.info("Spec not found for flowId {} due to deletion by active dagManager host due to exception {}",
+ flowId, e.getMessage());
+ // TODO: mark this failure if there are other valid cases of this exception
+ // launchSubmissionMetricProxy.markFailure();
Review Comment:
It's speculative, I think there could be a case where a SpecNotFoundException occurs in a case _not_ due to deletion by active DM if something goes wrong at a different point in the system (mistaken deletion NOT from the deleteFlowSpecIfAdhoc code path). It's hard to definitively say all of these errors are due to an active DM deleting it but I would expect them all to be.
##########
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java:
##########
@@ -321,8 +321,10 @@ protected void submitFlowToDagManagerHelper(DagActionStore.DagAction dagAction,
launchSubmissionMetricProxy.markFailure();
return;
} catch (SpecNotFoundException e) {
- log.warn("Spec not found for flowId {} due to exception {}", flowId, e.getMessage());
- launchSubmissionMetricProxy.markFailure();
+ log.info("Spec not found for flowId {} due to deletion by active dagManager host due to exception {}",
+ flowId, e.getMessage());
+ // TODO: mark this failure if there are other valid cases of this exception
+ // launchSubmissionMetricProxy.markFailure();
Review Comment:
It's speculative, I think there could be a case where a SpecNotFoundException occurs in a case _not_ due to deletion by active DM if something goes wrong at a different point in the system (mistaken deletion NOT from the deleteFlowSpecIfAdhoc code path). It's hard to definitively say all of these errors are due to an active DM deleting it but I would expect them all to be.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2107] Delete adhoc flowSpecs from flowCatalog bug fix [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on PR #3996:
URL: https://github.com/apache/gobblin/pull/3996#issuecomment-2221204478
> Are we not including this function in the DagProcEngine as well? Or is that out of scope
Good call out. I'm going to put that as out of scope, @arjun4084346 or I will add this in separate PR for `DagProcessingEngine`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2107] Delete adhoc flowSpecs from flowCatalog bug fix [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3996:
URL: https://github.com/apache/gobblin/pull/3996#discussion_r1671460273
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -271,14 +275,25 @@ public void orchestrate(Spec spec, Properties jobProps, long triggerTimestampMil
Instrumented.updateTimer(this.flowOrchestrationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
}
- public void submitFlowToDagManager(FlowSpec flowSpec) throws IOException, InterruptedException {
- Optional<Dag<JobExecutionPlan>> optionalJobExecutionPlanDag =
- this.flowCompilationValidationHelper.createExecutionPlanIfValid(flowSpec);
- if (optionalJobExecutionPlanDag.isPresent()) {
- submitFlowToDagManager(flowSpec, optionalJobExecutionPlanDag.get());
- } else {
- _log.warn("Flow: {} submitted to dagManager failed to compile and produce a job execution plan dag", flowSpec);
- Instrumented.markMeter(this.flowOrchestrationFailedMeter);
+ /**
+ * Method that accepts a flowSpec that it compiles before forwarding to the DagManagerfor execution. It's meant to be
+ * called by {@link org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor}
+ * @param flowSpec
+ * @throws IOException
+ * @throws InterruptedException
Review Comment:
suggest to simplify to:
```
* Compiles the provided {@link FlowSpec} into a {@link Dag<JobExecutionPlan>} it forwards to the {@link DagManager} for execution. It's meant to be
* called by {@link org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor}
```
tip: avoid "Method that... " or "Class that..." or "Code to..." in javadoc. since that could be said of *every* method/class, let's just take it as an implied given.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2107] Delete adhoc flowSpecs from flowCatalog bug fix [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on PR #3996:
URL: https://github.com/apache/gobblin/pull/3996#issuecomment-2219154425
> [I pointed out](https://github.com/apache/gobblin/pull/3944/files#r1591517872) in my earlier PR (you linked to) that we're missing test cases. I had added ones relating to my change. what would it take to exercise any of the bug fixes that motivate these latest mods?
I added a unit test `testDeleteFlowSpecCalledForMultiActivePath` that validates that the `compileAndSubmitFlowToDagManager` used by the `DagActionStoreChangeMonitor` attempts deleting adhoc `flowSpecs` from the catalog. Note that this test does not definitively ensure that this method is called by the `DagActionStoreChangeMonitor` which proves more complex but verifies if that method is used then the flow will be deleted after compilation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2107] Delete adhoc flowSpecs from flowCatalog bug fix [gobblin]
Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo merged PR #3996:
URL: https://github.com/apache/gobblin/pull/3996
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2107] Delete adhoc flowSpecs from flowCatalog bug fix [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3996:
URL: https://github.com/apache/gobblin/pull/3996#discussion_r1670930350
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -289,7 +293,7 @@ public void submitFlowToDagManager(FlowSpec flowSpec, Dag<JobExecutionPlan> jobE
Note that the responsibility of the multi-active scheduler mode ends after this method is completed AND the
consumption of a launch type event is committed to the consumer.
*/
- this.dagManager.addDag(jobExecutionPlanDag, true, true);
+ this.dagManager.addDagAndRemoveAdhocFlowSpec(flowSpec, jobExecutionPlanDag, true, true);
Review Comment:
now the attempt to remove an ad hoc spec would occur twice, since it's also on line 267.
I believe your aim is to have removal also happen in the path from [DagActionStoreChangeMonitor](https://github.com/apache/gobblin/blob/4721d086f78553cab28e46bf406787df6cc2dea8/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java#L318), correct?
couldn't that run into the same issue, where if the flow compilation validation helper fails (see L281 directly above), so the call to this new combo addDag+removeAdHoc would be skipped?
this new PR definitely shows me I missed the `DagActionStoreChangeMonitor` case in my recent PR (sorry to see). the crux there was to disentangle adding the dag from removing the ad hoc spec. the latter MUST happen, suggesting it belongs inside a `finally`, whereas the former is only best-effort. I would imagine you'll want the same approach here, in the change monitor code path I failed to notice
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2107] Delete adhoc flowSpecs from flowCatalog bug fix [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3996:
URL: https://github.com/apache/gobblin/pull/3996#discussion_r1670930350
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -289,7 +293,7 @@ public void submitFlowToDagManager(FlowSpec flowSpec, Dag<JobExecutionPlan> jobE
Note that the responsibility of the multi-active scheduler mode ends after this method is completed AND the
consumption of a launch type event is committed to the consumer.
*/
- this.dagManager.addDag(jobExecutionPlanDag, true, true);
+ this.dagManager.addDagAndRemoveAdhocFlowSpec(flowSpec, jobExecutionPlanDag, true, true);
Review Comment:
now the attempt to remove an ad hoc spec would occur twice, since it's also on line 267.
I believe your aim is to have removal also happen in the path from [DagActionStoreChangeMonitor](https://github.com/apache/gobblin/blob/4721d086f78553cab28e46bf406787df6cc2dea8/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java#L318), correct?
couldn't that run into the same issue, where if the flow compilation validation helper fails (see L281 directly above), so the call to this new combo addDag+removeAdHoc would be skipped?
this new PR definitely shows me I missed the `DagActionStoreChangeMonitor` case in my recent PR (sorry to see). the crux there was to disentangle adding the dag from removing the ad hoc spec. the latter MUST happen, suggesting it belongs inside a `finally`, whereas the former is only best-effort.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2107] Delete adhoc flowSpecs from flowCatalog bug fix [gobblin]
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3996:
URL: https://github.com/apache/gobblin/pull/3996#discussion_r1670930350
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java:
##########
@@ -289,7 +293,7 @@ public void submitFlowToDagManager(FlowSpec flowSpec, Dag<JobExecutionPlan> jobE
Note that the responsibility of the multi-active scheduler mode ends after this method is completed AND the
consumption of a launch type event is committed to the consumer.
*/
- this.dagManager.addDag(jobExecutionPlanDag, true, true);
+ this.dagManager.addDagAndRemoveAdhocFlowSpec(flowSpec, jobExecutionPlanDag, true, true);
Review Comment:
now the attempt to remove an ad hoc spec would occur twice, since it's also on line 267.
I believe your aim is to have removal also happen in the path from [DagActionStoreChangeMonitor](https://github.com/apache/gobblin/blob/4721d086f78553cab28e46bf406787df6cc2dea8/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java#L318), correct?
couldn't that run into the same issue, where if the flow compilation validation helper fails (see L281 directly above), so the call to this new combo addDag+removeAdHoc would be skipped?
this new PR helps me realize I had missed the `DagActionStoreChangeMonitor` case in my recent PR. the crux of that one was to divide adding the dag from removing the ad hoc spec. the latter MUST happen, suggesting it belongs inside a `finally`, whereas the former is only best-effort.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org