You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gobblin.apache.org by zi...@apache.org on 2023/03/11 01:11:34 UTC
[gobblin] branch master updated: fix add spec and actual number flows scheduled count (#3660)
This is an automated email from the ASF dual-hosted git repository.
zihanli58 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/gobblin.git
The following commit(s) were added to refs/heads/master by this push:
new d6b5c7920 fix add spec and actual number flows scheduled count (#3660)
d6b5c7920 is described below
commit d6b5c79205133c7ef3898fb0b04339a57b33b36c
Author: umustafi <um...@gmail.com>
AuthorDate: Fri Mar 10 17:11:28 2023 -0800
fix add spec and actual number flows scheduled count (#3660)
Co-authored-by: Urmi Mustafi <um...@umustafi-mn1.linkedin.biz>
---
.../scheduler/GobblinServiceJobScheduler.java | 23 +++++++++++++---------
1 file changed, 14 insertions(+), 9 deletions(-)
diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java
index 53713fe35..5ee5f9789 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java
@@ -247,8 +247,9 @@ public class GobblinServiceJobScheduler extends JobScheduler implements SpecCata
}
}
- /** Check that a spec should be scheduled and if it is, modify the spec of an adhoc flow before adding to scheduler*/
- private void addSpecHelperMethod(Spec spec) {
+ /** Return true if a spec should be scheduled and if it is, modify the spec of an adhoc flow before adding to
+ * scheduler. Return false otherwise. */
+ private boolean addSpecHelperMethod(Spec spec) {
// Adhoc flows will not have any job schedule key, but we should schedule them
if (spec instanceof FlowSpec) {
FlowSpec flowSpec = (FlowSpec) spec;
@@ -261,10 +262,12 @@ public class GobblinServiceJobScheduler extends JobScheduler implements SpecCata
} else {
onAddSpec(spec);
}
+ return true;
}
}else {
_log.debug("Not scheduling spec {} during startup as next job to schedule is outside of threshold.", spec);
}
+ return false;
}
/**
@@ -342,9 +345,10 @@ public class GobblinServiceJobScheduler extends JobScheduler implements SpecCata
while (batchOfSpecsIterator.hasNext()) {
Spec spec = batchOfSpecsIterator.next();
try {
- addSpecHelperMethod(spec);
- totalAddSpecTime += this.eachCompleteAddSpecValue; // this is updated by each call to onAddSpec
- actualNumFlowsScheduled += 1;
+ if (addSpecHelperMethod(spec)) {
+ totalAddSpecTime += this.eachCompleteAddSpecValue; // this is updated by each call to onAddSpec
+ actualNumFlowsScheduled += 1;
+ }
} catch (Exception e) {
// If there is an uncaught error thrown during compilation, log it and continue adding flows
_log.error("Could not schedule spec {} from flowCatalog due to ", spec, e);
@@ -369,12 +373,13 @@ public class GobblinServiceJobScheduler extends JobScheduler implements SpecCata
Spec spec = this.flowCatalog.get().getSpecWrapper(uri);
this.individualGetSpecSpeedValue = System.nanoTime() - individualGetSpecStartTime;
totalGetTime += this.individualGetSpecSpeedValue;
- addSpecHelperMethod(spec);
- totalAddSpecTime += this.eachCompleteAddSpecValue; // this is updated by each call to onAddSpec
- actualNumFlowsScheduled += 1;
+ if (addSpecHelperMethod(spec)) {
+ totalAddSpecTime += this.eachCompleteAddSpecValue; // this is updated by each call to onAddSpec
+ actualNumFlowsScheduled += 1;
+ }
} catch (Exception e) {
// If there is an uncaught error thrown during compilation, log it and continue adding flows
- _log.error("Could not schedule spec uri {} from flowCatalog due to ", uri, e);
+ _log.error("Could not schedule spec uri {} from flowCatalog due to {}", uri, e);
}
}
// Reset value after its last value to get an accurate reading