[
https://issues.apache.org/jira/browse/GOBBLIN-2151?focusedWorklogId=933897&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-933897
]
ASF GitHub Bot logged work on GOBBLIN-2151:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 10/Sep/24 01:56
Start Date: 10/Sep/24 01:56
Worklog Time Spent: 10m
Work Description: arjun4084346 commented on code in PR #4050:
URL: https://github.com/apache/gobblin/pull/4050#discussion_r1751140781
##########
gobblin-service/src/main/java/org/apache/gobblin/service/modules/utils/FlowCompilationValidationHelper.java:
##########
@@ -187,9 +195,65 @@ public Optional<Dag<JobExecutionPlan>>
validateAndHandleConcurrentExecution(Conf
* @param allowConcurrentExecution
* @return true if the {@link FlowSpec} allows concurrent executions or if
no other instance of the flow is currently RUNNING.
*/
- private boolean isExecutionPermitted(FlowStatusGenerator
flowStatusGenerator, String flowGroup, String flowName,
- boolean allowConcurrentExecution, long flowExecutionId) {
- return allowConcurrentExecution ||
!flowStatusGenerator.isFlowRunning(flowName, flowGroup, flowExecutionId);
+ private boolean isExecutionPermitted(String flowGroup, String flowName,
boolean allowConcurrentExecution)
+ throws IOException {
+ return allowConcurrentExecution || !isFlowRunning(flowGroup, flowName,
dagManagementStateStore);
+ }
+
+ /**
+ * Returns true if any previous execution for the flow determined by the
provided flowGroup, flowName is running.
+ * We ignore the execution that has the provided flowExecutionId. We also
ignore the flows that are running beyond
+ * the job start deadline and flow finish deadline.
+ */
+ @VisibleForTesting
+ static boolean isFlowRunning(String flowGroup, String flowName,
DagManagementStateStore dagManagementStateStore)
+ throws IOException {
+ List<FlowStatus> flowStatusList =
dagManagementStateStore.getAllFlowStatusesForFlow(flowGroup, flowName);
+
+ if (flowStatusList == null || flowStatusList.isEmpty()) {
+ return false;
+ }
+
+ for (FlowStatus flowStatus : flowStatusList) {
+ ExecutionStatus flowExecutionStatus =
flowStatus.getFlowExecutionStatus();
+ log.debug("Verifying if {} is running...", flowStatus);
+
+ if
(FlowStatusGenerator.FINISHED_STATUSES.contains(flowExecutionStatus.name())) {
+ // ignore finished entries
+ } else if (flowExecutionStatus == COMPILED || flowExecutionStatus ==
PENDING
+ || flowExecutionStatus == PENDING_RESUME || flowExecutionStatus ==
RUNNING) {
+ // these are the only four non-terminal statuses that a flow can have.
jobs have two more non-terminal statuses
+ // ORCHESTRATED and PENDING_RETRY
+ Dag.DagId dagIdOfOldExecution = new Dag.DagId(flowGroup, flowName,
flowStatus.getFlowExecutionId());
+ java.util.Optional<Dag<JobExecutionPlan>> dag =
dagManagementStateStore.getDag(dagIdOfOldExecution);
+
+ if (!dag.isPresent()) {
+ log.error("Dag is finished and cleaned up, job status monitor
somehow did not receive/update the flow status. Ignoring it here...");
+ continue;
+ }
+
+ Dag.DagNode<JobExecutionPlan> dagNode = dag.get().getNodes().get(0);
+ long flowStartTime = DagUtils.getFlowStartTime(dagNode);
+ long jobStartDeadline =
+ DagUtils.getJobStartDeadline(dagNode,
DagProcessingEngine.getDefaultJobStartDeadlineTimeMillis());
+ long flowFinishDeadline = DagUtils.getFlowFinishDeadline(dagNode);
+ if ((flowExecutionStatus == COMPILED || flowExecutionStatus ==
PENDING)
+ && System.currentTimeMillis() < flowStartTime +
jobStartDeadline
+ || (flowExecutionStatus == RUNNING || flowExecutionStatus ==
PENDING_RESUME)
+ && System.currentTimeMillis() < flowStartTime +
flowFinishDeadline) {
+ log.info("{} is still running. Found a dag for this, flowStartTime
{}, jobStartDeadline {}, flowFinishDeadline {}",
+ flowStatus, flowStartTime, jobStartDeadline,
flowFinishDeadline);
+ return true;
+ } else {
+ log.warn("Dag {} is still running beyond deadline! flowStartTime
{}, jobStartDeadline {}, flowFinishDeadline {}",
+ dag, flowStartTime, jobStartDeadline, flowFinishDeadline);
+ }
+ } else {
+ log.error("Unknown status {}", flowExecutionStatus);
Review Comment:
imo, UNKNOWN status itself is an indication of something wrong.
Issue Time Tracking
-------------------
Worklog Id: (was: 933897)
Time Spent: 1h 10m (was: 1h)
> ignore flows that are running beyond job start and flow finish deadline
> -----------------------------------------------------------------------
>
> Key: GOBBLIN-2151
> URL: https://issues.apache.org/jira/browse/GOBBLIN-2151
> Project: Apache Gobblin
> Issue Type: Improvement
> Reporter: Arjun Singh Bora
> Priority: Major
> Time Spent: 1h 10m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
