[
https://issues.apache.org/jira/browse/GOBBLIN-2151?focusedWorklogId=933894&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-933894
]
ASF GitHub Bot logged work on GOBBLIN-2151:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 10/Sep/24 01:21
Start Date: 10/Sep/24 01:21
Worklog Time Spent: 10m
Work Description: phet commented on code in PR #4050:
URL: https://github.com/apache/gobblin/pull/4050#discussion_r1751095651
##########
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:
`ORCHESTRATED` isn't a final status is it?
also, how to handle `$UNKNOWN`? shall we say it's NOT running?
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/utils/FlowCompilationValidationHelperTest.java:
##########
@@ -66,4 +100,127 @@ public void testSkipAddingFlowExecutionIdWhenPresent() {
Assert.assertEquals(flowMetadata.get(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD),
existingFlowExecutionId);
}
+
+ @Test
+ public void testConcurrentFlowPreviousFlowWithNonTerminalStatusWithNoDag()
throws IOException {
+ List<FlowStatus> list = new ArrayList<>();
+ String flowGroup = "fg";
+ String flowName = "fn";
+ long previousFlowExecutionId = 12345L;
+ JobStatus jobStatus =
JobStatus.builder().flowGroup(flowGroup).flowName(flowName).flowExecutionId(previousFlowExecutionId)
+
.jobName(JobStatusRetriever.NA_KEY).jobGroup(JobStatusRetriever.NA_KEY).eventName(ExecutionStatus.COMPILED.name()).build();
+ Iterator<JobStatus> jobStatusIterator =
Lists.newArrayList(jobStatus).iterator();
+ list.add(new FlowStatus(flowName, flowGroup, previousFlowExecutionId,
jobStatusIterator, ExecutionStatus.COMPILED));
+ when(this.dagManagementStateStore.getAllFlowStatusesForFlow(anyString(),
anyString())).thenReturn(list);
+
+
Assert.assertFalse(FlowCompilationValidationHelper.isFlowRunning(flowGroup,
flowName, this.dagManagementStateStore));
+ }
+
+ @Test
+ public void
testConcurrentFlowPreviousFlowWithNonTerminalStatusRunningBeyondJobStartDeadline()
+ throws IOException, URISyntaxException {
+ List<FlowStatus> list = new ArrayList<>();
+ String flowGroup = "fg";
+ String flowName = "fn";
+ long previousFlowExecutionId = 12345L;
+ long jobStartDeadline = 10L;
+ long flowStartTime = System.currentTimeMillis() - jobStartDeadline - 1;
+ JobStatus jobStatus =
JobStatus.builder().flowGroup(flowGroup).flowName(flowName).flowExecutionId(previousFlowExecutionId)
+
.jobName(JobStatusRetriever.NA_KEY).jobGroup(JobStatusRetriever.NA_KEY).eventName(ExecutionStatus.PENDING.name()).build();
+ Iterator<JobStatus> jobStatusIterator =
Lists.newArrayList(jobStatus).iterator();
+ list.add(new FlowStatus(flowName, flowGroup, previousFlowExecutionId,
jobStatusIterator, ExecutionStatus.PENDING));
+ when(this.dagManagementStateStore.getAllFlowStatusesForFlow(anyString(),
anyString())).thenReturn(list);
+ Dag<JobExecutionPlan> dag = DagManagerTest.buildDag("1",
previousFlowExecutionId,
+ DagProcessingEngine.FailureOption.FINISH_ALL_POSSIBLE.name(), 5,
"user5", ConfigFactory.empty()
+ .withValue(ConfigurationKeys.FLOW_GROUP_KEY,
ConfigValueFactory.fromAnyRef(flowGroup))
+ .withValue(ConfigurationKeys.FLOW_NAME_KEY,
ConfigValueFactory.fromAnyRef(flowName))
+ .withValue(ConfigurationKeys.GOBBLIN_JOB_START_DEADLINE_TIME_UNIT,
ConfigValueFactory.fromAnyRef(TimeUnit.MILLISECONDS.name()))
+ .withValue(ConfigurationKeys.GOBBLIN_JOB_START_DEADLINE_TIME,
ConfigValueFactory.fromAnyRef(jobStartDeadline))
+ .withValue(ConfigurationKeys.SPECEXECUTOR_INSTANCE_URI_KEY,
ConfigValueFactory.fromAnyRef(
+ MySqlDagManagementStateStoreTest.TEST_SPEC_EXECUTOR_URI)));
+ dag.getNodes().forEach(node ->
node.getValue().setFlowStartTime(flowStartTime));
+ this.dagManagementStateStore.addDag(dag);
+
+
Assert.assertFalse(FlowCompilationValidationHelper.isFlowRunning(flowGroup,
flowName, this.dagManagementStateStore));
+ }
+
+ @Test
+ public void
testConcurrentFlowPreviousFlowWithNonTerminalStatusRunningBeyondFlowFinishDeadline()
+ throws IOException, URISyntaxException {
+ List<FlowStatus> list = new ArrayList<>();
+ String flowGroup = "fg";
+ String flowName = "fn";
+ long previousFlowExecutionId = 12345L;
+ long flowFinishDeadline = 30L;
+ long flowStartTime = System.currentTimeMillis() - flowFinishDeadline - 1;
+ JobStatus jobStatus =
JobStatus.builder().flowGroup(flowGroup).flowName(flowName).flowExecutionId(previousFlowExecutionId)
+
.jobName(JobStatusRetriever.NA_KEY).jobGroup(JobStatusRetriever.NA_KEY).eventName(ExecutionStatus.PENDING_RESUME.name())
+ .build();
+ Iterator<JobStatus> jobStatusIterator =
Lists.newArrayList(jobStatus).iterator();
+ list.add(new FlowStatus(flowName, flowGroup, previousFlowExecutionId,
jobStatusIterator, ExecutionStatus.PENDING_RESUME));
+ when(this.dagManagementStateStore.getAllFlowStatusesForFlow(anyString(),
anyString())).thenReturn(list);
+ Dag<JobExecutionPlan> dag = DagManagerTest.buildDag("1",
previousFlowExecutionId,
+ DagProcessingEngine.FailureOption.FINISH_ALL_POSSIBLE.name(), 5,
"user5", ConfigFactory.empty()
+ .withValue(ConfigurationKeys.FLOW_GROUP_KEY,
ConfigValueFactory.fromAnyRef(flowGroup))
+ .withValue(ConfigurationKeys.FLOW_NAME_KEY,
ConfigValueFactory.fromAnyRef(flowName))
+
.withValue(ConfigurationKeys.GOBBLIN_FLOW_FINISH_DEADLINE_TIME_UNIT,
ConfigValueFactory.fromAnyRef(TimeUnit.MILLISECONDS.name()))
+ .withValue(ConfigurationKeys.GOBBLIN_FLOW_FINISH_DEADLINE_TIME,
ConfigValueFactory.fromAnyRef(flowFinishDeadline))
+ .withValue(ConfigurationKeys.SPECEXECUTOR_INSTANCE_URI_KEY,
ConfigValueFactory.fromAnyRef(
+ MySqlDagManagementStateStoreTest.TEST_SPEC_EXECUTOR_URI)));
+ dag.getNodes().forEach(node ->
node.getValue().setFlowStartTime(flowStartTime));
+ this.dagManagementStateStore.addDag(dag);
+
+
Assert.assertFalse(FlowCompilationValidationHelper.isFlowRunning(flowGroup,
flowName, this.dagManagementStateStore));
+ }
+
+ @Test
+ public void
testConcurrentFlowPreviousFlowWithNonTerminalStatusRunningWithinFlowFinishDeadline()
+ throws IOException, URISyntaxException {
+ List<FlowStatus> list = new ArrayList<>();
+ String flowGroup = "fg";
+ String flowName = "fn";
+ long previousFlowExecutionId = 12345L;
+ long flowFinishDeadline = 10000L;
+ long flowStartTime = System.currentTimeMillis(); // giving test
flowFinishDeadline to finish
+ JobStatus jobStatus =
JobStatus.builder().flowGroup(flowGroup).flowName(flowName).flowExecutionId(previousFlowExecutionId)
+
.jobName(JobStatusRetriever.NA_KEY).jobGroup(JobStatusRetriever.NA_KEY).eventName(ExecutionStatus.RUNNING.name())
+ .build();
+ Iterator<JobStatus> jobStatusIterator =
Lists.newArrayList(jobStatus).iterator();
+ list.add(new FlowStatus(flowName, flowGroup, previousFlowExecutionId,
jobStatusIterator, ExecutionStatus.RUNNING));
+ when(this.dagManagementStateStore.getAllFlowStatusesForFlow(anyString(),
anyString())).thenReturn(list);
+ Dag<JobExecutionPlan> dag = DagManagerTest.buildDag("1",
previousFlowExecutionId,
+ DagProcessingEngine.FailureOption.FINISH_ALL_POSSIBLE.name(), 5,
"user5", ConfigFactory.empty()
+ .withValue(ConfigurationKeys.FLOW_GROUP_KEY,
ConfigValueFactory.fromAnyRef(flowGroup))
+ .withValue(ConfigurationKeys.FLOW_NAME_KEY,
ConfigValueFactory.fromAnyRef(flowName))
+
.withValue(ConfigurationKeys.GOBBLIN_FLOW_FINISH_DEADLINE_TIME_UNIT,
ConfigValueFactory.fromAnyRef(TimeUnit.MILLISECONDS.name()))
+ .withValue(ConfigurationKeys.GOBBLIN_FLOW_FINISH_DEADLINE_TIME,
ConfigValueFactory.fromAnyRef(flowFinishDeadline))
+ .withValue(ConfigurationKeys.SPECEXECUTOR_INSTANCE_URI_KEY,
ConfigValueFactory.fromAnyRef(
+ MySqlDagManagementStateStoreTest.TEST_SPEC_EXECUTOR_URI)));
+ dag.getNodes().forEach(node ->
node.getValue().setFlowStartTime(flowStartTime));
+ this.dagManagementStateStore.addDag(dag);
Review Comment:
this DMSS mock config is so extensive that it really overshadows the intent
of these tests. how about abstracting into:
```
public static void insertFlowIntoDMSSMock(String flowGroup, String flowName,
long flowExecId, ExecutionStatus currFlowAndJobStatus, Config additionalConfigs)
```
NOTES:
* choose random username
* set flowStartTime == flowExecId
* use additionalConfigs as a fallback after adding: FLOW_GROUP_KEY,
FLOW_NAME_KEY, and SPECEXECUTOR_INSTANCE_URI_KEY
* always set GOBBLIN_FLOW_FINISH_DEADLINE_TIME_UNIT and
GOBBLIN_JOB_START_DEADLINE_TIME_UNIT to TimeUnit.MILLISECONDS... even if it may
not actually be necessary
with that, the setup and the assertion here are far simpler and easier to
grasp:
```
insertFlowIntoDMSSMock(flowGroup, flowName, previousFlowExecutionId,
ExecutionStatus.RUNNING,
ConfigFactory.empty().
withValue(ConfigurationKeys.GOBBLIN_FLOW_FINISH_DEADLINE_TIME,
ConfigValueFactory.fromAnyRef(flowFinishDeadline)));
Assert.assertTrue(FlowCompilationValidationHelper.isFlowRunning(flowGroup,
flowName, this.dagManagementStateStore));
```
Issue Time Tracking
-------------------
Worklog Id: (was: 933894)
Time Spent: 50m (was: 40m)
> 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: 50m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
