You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by GitBox <gi...@apache.org> on 2022/06/13 18:36:05 UTC
[GitHub] [gobblin] arjun4084346 commented on a diff in pull request #3520: [GOBBLIN-1662] Fix running counts for retried flows
arjun4084346 commented on code in PR #3520:
URL: https://github.com/apache/gobblin/pull/3520#discussion_r896021375
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -963,7 +963,8 @@ private void submitJob(DagNode<JobExecutionPlan> dagNode) {
// By this point the quota is allocated, so it's imperative to increment as missing would introduce the potential to decrement below zero upon quota release.
// Quota release is guaranteed, despite failure, because exception handling within would mark the job FAILED.
// When the ensuing kafka message spurs DagManager processing, the quota is released and the counts decremented
- if (this.metricContext != null) {
+ // Ensure that we do not double increment for flows that are retried
+ if (this.metricContext != null && dagNode.getValue().getCurrentAttempts() == 1) {
Review Comment:
Should we add this check while decrementing ?
where we have ```getRunningJobsCounter(dagNode).dec()```
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java:
##########
@@ -963,7 +963,8 @@ private void submitJob(DagNode<JobExecutionPlan> dagNode) {
// By this point the quota is allocated, so it's imperative to increment as missing would introduce the potential to decrement below zero upon quota release.
// Quota release is guaranteed, despite failure, because exception handling within would mark the job FAILED.
// When the ensuing kafka message spurs DagManager processing, the quota is released and the counts decremented
- if (this.metricContext != null) {
+ // Ensure that we do not double increment for flows that are retried
+ if (this.metricContext != null && dagNode.getValue().getCurrentAttempts() == 1) {
Review Comment:
Should we add this check while decrementing also?
where we have ```getRunningJobsCounter(dagNode).dec()```
--
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