[
https://issues.apache.org/jira/browse/GOBBLIN-1552?focusedWorklogId=660125&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-660125
]
ASF GitHub Bot logged work on GOBBLIN-1552:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 05/Oct/21 06:09
Start Date: 05/Oct/21 06:09
Worklog Time Spent: 10m
Work Description: phet commented on a change in pull request #3403:
URL: https://github.com/apache/gobblin/pull/3403#discussion_r721891280
##########
File path:
gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowExecutionResourceLocalHandler.java
##########
@@ -87,6 +87,8 @@ public FlowExecution get(ComplexResourceKey<FlowStatusId,
EmptyRecord> key) {
getLatestFlowGroupStatusesFromGenerator(flowGroup, countPerFlow, tag,
this.flowStatusGenerator);
if (flowStatuses != null) {
+ // todo: flow end time will be incorrect when dag manager is not used
Review comment:
is it correct that it would be `0L`? if so, just wondering whether
problematic to break the reasonable presumption that endTime > startTime. if
is, possibly use MAX_LONG?
also, musing, could we approximate with greatest end time among the job
statuses here (even though none `isFlowStatus`)--or do those not have the end
time even set?
##########
File path:
gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java
##########
@@ -218,4 +228,42 @@ public static boolean
isFlowStatus(org.apache.gobblin.service.monitoring.JobStat
return jobStatus.getJobName() != null && jobStatus.getJobGroup() != null
&& jobStatus.getJobName().equals(JobStatusRetriever.NA_KEY) &&
jobStatus.getJobGroup().equals(JobStatusRetriever.NA_KEY);
}
+
+ public static ExecutionStatus getFlowStatusFromJobStatuses(boolean
dagManagerEnabled, Iterator<JobStatus> jobStatusIterator) {
+ ExecutionStatus flowExecutionStatus = ExecutionStatus.$UNKNOWN;
+
+ if (dagManagerEnabled) {
+ while (jobStatusIterator.hasNext()) {
+ JobStatus jobStatus = jobStatusIterator.next();
+ // Check if this is the flow status instead of a single job status
+ if (JobStatusRetriever.isFlowStatus(jobStatus)) {
+ flowExecutionStatus =
ExecutionStatus.valueOf(jobStatus.getEventName());
+ }
+ }
+ } else {
+ Set<ExecutionStatus> jobStatuses = new HashSet<>();
+ while (jobStatusIterator.hasNext()) {
+ JobStatus jobStatus = jobStatusIterator.next();
+ // because in absence of DagManager we do not get all flow level
events, we will ignore the flow level events
+ // we actually get and purely calculate flow status based on flow
statuses.
Review comment:
nice comment--(future) maintainers will sing your praises ;p
##########
File path:
gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/FlowStatus.java
##########
@@ -39,4 +40,5 @@
private final long flowExecutionId;
@ToString.Exclude // (to avoid side-effecting exhaustion of `Iterator`)
Review comment:
FYI, I do prefer this approach, here -
https://github.com/apache/gobblin/pull/3402 (of replacing `Iterator` w/ `List`)
...but I understand if you'd like to introduce them as separate changes.
##########
File path:
gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java
##########
@@ -173,13 +179,17 @@ protected final long getJobExecutionId(State jobState) {
}
protected List<FlowStatus>
asFlowStatuses(List<FlowExecutionJobStateGrouping> flowExecutionGroupings) {
- return flowExecutionGroupings.stream().map(exec ->
- new FlowStatus(exec.getFlowName(), exec.getFlowGroup(),
exec.getFlowExecutionId(),
- asJobStatuses(exec.getJobStates().stream().sorted(
- // rationalized order, to facilitate test assertions
-
Comparator.comparing(this::getJobGroup).thenComparing(this::getJobName).thenComparing(this::getJobExecutionId)
- ).collect(Collectors.toList()))))
- .collect(Collectors.toList());
+ return flowExecutionGroupings.stream().map(exec -> {
+ List<JobStatus> jobStatuses =
ImmutableList.copyOf(asJobStatuses(exec.getJobStates().stream().sorted(
+ // rationalized order, to facilitate test assertions
+
Comparator.comparing(this::getJobGroup).thenComparing(this::getJobName).thenComparing(this::getJobExecutionId)
+ ).collect(Collectors.toList())));
+ Iterator<JobStatus> jobStatusIterator = jobStatuses.iterator();
Review comment:
repeat suggestion: inline `jobStatuses.iterator()`, if you choose
##########
File path:
gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/FlowStatusGenerator.java
##########
@@ -127,10 +115,15 @@ private String getExecutionStatus(Iterator<JobStatus>
jobStatusIterator) {
* list only contains jobs matching the tag.
*/
public FlowStatus getFlowStatus(String flowName, String flowGroup, long
flowExecutionId, String tag) {
- Iterator<JobStatus> jobStatusIterator =
retainStatusOfAnyFlowOrJobMatchingTag(
- jobStatusRetriever.getJobStatusesForFlowExecution(flowName, flowGroup,
flowExecutionId), tag);
-
- return jobStatusIterator.hasNext() ? new FlowStatus(flowName, flowGroup,
flowExecutionId, jobStatusIterator) : null;
+ List<JobStatus> jobStatuses =
ImmutableList.copyOf(retainStatusOfAnyFlowOrJobMatchingTag(
+ jobStatusRetriever.getJobStatusesForFlowExecution(flowName, flowGroup,
flowExecutionId), tag));
+ Iterator<JobStatus> jobStatusIterator = jobStatuses.iterator();
Review comment:
minor: I wouldn't actually create the binding, but just inline
`jobStatuses.iterator()` in both places... but up to you
##########
File path:
gobblin-runtime/src/test/java/org/apache/gobblin/service/monitoring/FlowStatusGeneratorTest.java
##########
@@ -134,8 +143,9 @@ public void testGetFlowStatusesAcrossGroup() {
Arrays.asList(f0jsmDep2)));
}
- private FlowStatus createFlowStatus(String flowGroup, String flowName, long
flowExecutionId, List<JobStatus> jobStatuses) {
- return new FlowStatus(flowName, flowGroup, flowExecutionId,
jobStatuses.iterator());
+ private FlowStatus createFlowStatus(String flowGroup, String flowName, long
flowExecutionId, List<JobStatus> jobStatuses, JobStatusRetriever
jobStatusRetriever) {
+ return new FlowStatus(flowName, flowGroup, flowExecutionId,
jobStatuses.iterator(),
+
jobStatusRetriever.getFlowStatusFromJobStatuses(jobStatusRetriever.dagManagerEnabled,
jobStatuses.iterator()));
Review comment:
probably better to call through the class,
`JobStatusRetriever.getFlowStatusFromJobStatuses`, and drop the final param
##########
File path:
gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java
##########
@@ -218,4 +228,42 @@ public static boolean
isFlowStatus(org.apache.gobblin.service.monitoring.JobStat
return jobStatus.getJobName() != null && jobStatus.getJobGroup() != null
&& jobStatus.getJobName().equals(JobStatusRetriever.NA_KEY) &&
jobStatus.getJobGroup().equals(JobStatusRetriever.NA_KEY);
}
+
+ public static ExecutionStatus getFlowStatusFromJobStatuses(boolean
dagManagerEnabled, Iterator<JobStatus> jobStatusIterator) {
Review comment:
no biggie... just thinking: you could name the flag semantically, such
as `shouldExpectFlowStatuses` or `relyOnFlowStatuses`, rather than what
circumstance (DagManager enablement), leads us to expect/rely on them or not.
##########
File path:
gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/JobStatusRetrieverTest.java
##########
@@ -103,8 +103,8 @@ public void testGetJobStatusesForFlowExecution() throws
IOException {
long flowExecutionId = 1234L;
addJobStatusToStateStore(flowExecutionId, JobStatusRetriever.NA_KEY,
ExecutionStatus.COMPILED.name());
- Iterator<JobStatus>
- jobStatusIterator =
this.jobStatusRetriever.getJobStatusesForFlowExecution(FLOW_NAME, FLOW_GROUP,
flowExecutionId);
+ List<JobStatus> jobStatuses =
ImmutableList.copyOf(this.jobStatusRetriever.getJobStatusesForFlowExecution(FLOW_NAME,
FLOW_GROUP, flowExecutionId));
Review comment:
could invoke statically now
##########
File path:
gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/FsJobStatusRetriever.java
##########
@@ -60,7 +62,8 @@
@Inject
public FsJobStatusRetriever(Config config, MultiContextIssueRepository
issueRepository) {
- super(issueRepository);
+ super(ConfigUtils.getBoolean(config,
ServiceConfigKeys.GOBBLIN_SERVICE_DAG_MANAGER_ENABLED_KEY,
+ ServiceConfigKeys.DEFAULT_GOBBLIN_SERVICE_DAG_MANAGER_ENABLED),
issueRepository);
Review comment:
could you raise the `Config` to an even higher level, so this too takes
the boolean param... or too difficult to do w/ DI / wiring logic?
##########
File path:
gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/MysqlJobStatusRetrieverTestWithoutDagManager.java
##########
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.monitoring;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Properties;
+
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.google.common.base.Strings;
+
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.metastore.MysqlJobStatusStateStore;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.runtime.troubleshooter.MultiContextIssueRepository;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.ServiceConfigKeys;
+
+import static org.mockito.Mockito.mock;
+
+
+public class MysqlJobStatusRetrieverTestWithoutDagManager extends
JobStatusRetrieverTest {
+ private MysqlJobStatusStateStore<State> dbJobStateStore;
+ private static final String TEST_USER = "testUser";
+ private static final String TEST_PASSWORD = "testPassword";
+
+ @BeforeClass
+ @Override
+ public void setUp() throws Exception {
+ ITestMetastoreDatabase testMetastoreDatabase =
TestMetastoreDatabaseFactory.get();
+ String jdbcUrl = testMetastoreDatabase.getJdbcUrl();
+
+ ConfigBuilder configBuilder = ConfigBuilder.create();
+
configBuilder.addPrimitive(MysqlJobStatusRetriever.MYSQL_JOB_STATUS_RETRIEVER_PREFIX
+ "." + ConfigurationKeys.STATE_STORE_DB_URL_KEY, jdbcUrl);
+
configBuilder.addPrimitive(MysqlJobStatusRetriever.MYSQL_JOB_STATUS_RETRIEVER_PREFIX
+ "." + ConfigurationKeys.STATE_STORE_DB_USER_KEY, TEST_USER);
+
configBuilder.addPrimitive(MysqlJobStatusRetriever.MYSQL_JOB_STATUS_RETRIEVER_PREFIX
+ "." + ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, TEST_PASSWORD);
+
+ this.jobStatusRetriever =
+ new MysqlJobStatusRetriever(configBuilder.build(),
mock(MultiContextIssueRepository.class));
+
configBuilder.addPrimitive(ServiceConfigKeys.GOBBLIN_SERVICE_DAG_MANAGER_ENABLED_KEY,
"true");
+ this.dbJobStateStore = ((MysqlJobStatusRetriever)
this.jobStatusRetriever).getStateStore();
+ cleanUpDir();
+ }
+
+ @Test
+ public void testGetJobStatusesForFlowExecution() throws IOException {
+ super.testGetJobStatusesForFlowExecution();
+ }
+
+ @Test (dependsOnMethods = "testGetJobStatusesForFlowExecution")
+ public void testJobTiming() throws Exception {
+ super.testJobTiming();
+ }
+
+ @Test (dependsOnMethods = "testJobTiming")
+ public void testOutOfOrderJobTimingEvents() throws IOException {
+ super.testOutOfOrderJobTimingEvents();
+ }
+
+ @Test (dependsOnMethods = "testJobTiming")
+ public void testGetJobStatusesForFlowExecution1() {
+ super.testGetJobStatusesForFlowExecution1();
+ }
+
+ @Test (dependsOnMethods = "testGetJobStatusesForFlowExecution1")
+ public void testGetLatestExecutionIdsForFlow() throws Exception {
+ super.testGetLatestExecutionIdsForFlow();
+ }
+
+ @Test (dependsOnMethods = "testGetLatestExecutionIdsForFlow")
+ public void testGetFlowStatusFromJobStatuses() throws Exception {
+ long flowExecutionId = 1237L;
+
+ addJobStatusToStateStore(flowExecutionId, JobStatusRetriever.NA_KEY,
ExecutionStatus.COMPILED.name());
+ Assert.assertEquals(ExecutionStatus.$UNKNOWN,
+
jobStatusRetriever.getFlowStatusFromJobStatuses(jobStatusRetriever.dagManagerEnabled,
jobStatusRetriever.getJobStatusesForFlowExecution(FLOW_NAME, FLOW_GROUP,
flowExecutionId)));
+
+ addJobStatusToStateStore(flowExecutionId, JobStatusRetriever.NA_KEY,
ExecutionStatus.ORCHESTRATED.name());
+ Assert.assertEquals(ExecutionStatus.$UNKNOWN,
+
jobStatusRetriever.getFlowStatusFromJobStatuses(jobStatusRetriever.dagManagerEnabled,
jobStatusRetriever.getJobStatusesForFlowExecution(FLOW_NAME, FLOW_GROUP,
flowExecutionId)));
+
+ addJobStatusToStateStore(flowExecutionId, MY_JOB_NAME_1,
ExecutionStatus.ORCHESTRATED.name(), JOB_ORCHESTRATED_TIME,
JOB_ORCHESTRATED_TIME);
+ Assert.assertEquals(ExecutionStatus.ORCHESTRATED,
+
jobStatusRetriever.getFlowStatusFromJobStatuses(jobStatusRetriever.dagManagerEnabled,
jobStatusRetriever.getJobStatusesForFlowExecution(FLOW_NAME, FLOW_GROUP,
flowExecutionId)));
+
+ addJobStatusToStateStore(flowExecutionId, JobStatusRetriever.NA_KEY,
ExecutionStatus.RUNNING.name());
+ Assert.assertEquals(ExecutionStatus.ORCHESTRATED,
+
jobStatusRetriever.getFlowStatusFromJobStatuses(jobStatusRetriever.dagManagerEnabled,
jobStatusRetriever.getJobStatusesForFlowExecution(FLOW_NAME, FLOW_GROUP,
flowExecutionId)));
+
+ addJobStatusToStateStore(flowExecutionId, MY_JOB_NAME_1,
ExecutionStatus.RUNNING.name(), JOB_ORCHESTRATED_TIME, JOB_ORCHESTRATED_TIME);
+ Assert.assertEquals(ExecutionStatus.RUNNING,
+
jobStatusRetriever.getFlowStatusFromJobStatuses(jobStatusRetriever.dagManagerEnabled,
jobStatusRetriever.getJobStatusesForFlowExecution(FLOW_NAME, FLOW_GROUP,
flowExecutionId)));
+
+ addJobStatusToStateStore(flowExecutionId, JobStatusRetriever.NA_KEY,
ExecutionStatus.COMPLETE.name(), JOB_ORCHESTRATED_TIME, JOB_ORCHESTRATED_TIME);
+ Assert.assertEquals(ExecutionStatus.RUNNING,
+
jobStatusRetriever.getFlowStatusFromJobStatuses(jobStatusRetriever.dagManagerEnabled,
jobStatusRetriever.getJobStatusesForFlowExecution(FLOW_NAME, FLOW_GROUP,
flowExecutionId)));
+
+ addJobStatusToStateStore(flowExecutionId, MY_JOB_NAME_1,
ExecutionStatus.COMPLETE.name());
+ Assert.assertEquals(ExecutionStatus.COMPLETE,
+
jobStatusRetriever.getFlowStatusFromJobStatuses(jobStatusRetriever.dagManagerEnabled,
jobStatusRetriever.getJobStatusesForFlowExecution(FLOW_NAME, FLOW_GROUP,
flowExecutionId)));
+ }
+
+ @Test
+ public void testMaxColumnName() throws Exception {
Review comment:
I don't initially see how this relates to the 'main/' changes above...
if it does, perhaps add a javadoc explanation about how this doesn't duplicate,
but instead tests a distinct code path.
##########
File path:
gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/MysqlJobStatusRetrieverTestWithoutDagManager.java
##########
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.monitoring;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Properties;
+
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.google.common.base.Strings;
+
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.metastore.MysqlJobStatusStateStore;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.runtime.troubleshooter.MultiContextIssueRepository;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.ServiceConfigKeys;
+
+import static org.mockito.Mockito.mock;
+
+
+public class MysqlJobStatusRetrieverTestWithoutDagManager extends
JobStatusRetrieverTest {
+ private MysqlJobStatusStateStore<State> dbJobStateStore;
+ private static final String TEST_USER = "testUser";
+ private static final String TEST_PASSWORD = "testPassword";
+
+ @BeforeClass
+ @Override
+ public void setUp() throws Exception {
+ ITestMetastoreDatabase testMetastoreDatabase =
TestMetastoreDatabaseFactory.get();
+ String jdbcUrl = testMetastoreDatabase.getJdbcUrl();
+
+ ConfigBuilder configBuilder = ConfigBuilder.create();
+
configBuilder.addPrimitive(MysqlJobStatusRetriever.MYSQL_JOB_STATUS_RETRIEVER_PREFIX
+ "." + ConfigurationKeys.STATE_STORE_DB_URL_KEY, jdbcUrl);
+
configBuilder.addPrimitive(MysqlJobStatusRetriever.MYSQL_JOB_STATUS_RETRIEVER_PREFIX
+ "." + ConfigurationKeys.STATE_STORE_DB_USER_KEY, TEST_USER);
+
configBuilder.addPrimitive(MysqlJobStatusRetriever.MYSQL_JOB_STATUS_RETRIEVER_PREFIX
+ "." + ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, TEST_PASSWORD);
+
+ this.jobStatusRetriever =
+ new MysqlJobStatusRetriever(configBuilder.build(),
mock(MultiContextIssueRepository.class));
+
configBuilder.addPrimitive(ServiceConfigKeys.GOBBLIN_SERVICE_DAG_MANAGER_ENABLED_KEY,
"true");
Review comment:
clearer to have line 62 precede line 61 (the `.build()`)... but change
`"true"` to `"false"`
##########
File path:
gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/MysqlJobStatusRetrieverTestWithoutDagManager.java
##########
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.monitoring;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Properties;
+
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.google.common.base.Strings;
+
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.metastore.MysqlJobStatusStateStore;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.runtime.troubleshooter.MultiContextIssueRepository;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.ServiceConfigKeys;
+
+import static org.mockito.Mockito.mock;
+
+
+public class MysqlJobStatusRetrieverTestWithoutDagManager extends
JobStatusRetrieverTest {
Review comment:
needs javadoc
##########
File path:
gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java
##########
@@ -218,4 +228,42 @@ public static boolean
isFlowStatus(org.apache.gobblin.service.monitoring.JobStat
return jobStatus.getJobName() != null && jobStatus.getJobGroup() != null
&& jobStatus.getJobName().equals(JobStatusRetriever.NA_KEY) &&
jobStatus.getJobGroup().equals(JobStatusRetriever.NA_KEY);
}
+
+ public static ExecutionStatus getFlowStatusFromJobStatuses(boolean
dagManagerEnabled, Iterator<JobStatus> jobStatusIterator) {
+ ExecutionStatus flowExecutionStatus = ExecutionStatus.$UNKNOWN;
+
+ if (dagManagerEnabled) {
+ while (jobStatusIterator.hasNext()) {
+ JobStatus jobStatus = jobStatusIterator.next();
+ // Check if this is the flow status instead of a single job status
+ if (JobStatusRetriever.isFlowStatus(jobStatus)) {
+ flowExecutionStatus =
ExecutionStatus.valueOf(jobStatus.getEventName());
+ }
+ }
+ } else {
+ Set<ExecutionStatus> jobStatuses = new HashSet<>();
+ while (jobStatusIterator.hasNext()) {
+ JobStatus jobStatus = jobStatusIterator.next();
+ // because in absence of DagManager we do not get all flow level
events, we will ignore the flow level events
+ // we actually get and purely calculate flow status based on flow
statuses.
+ if (!JobStatusRetriever.isFlowStatus(jobStatus)) {
+ jobStatuses.add(ExecutionStatus.valueOf(jobStatus.getEventName()));
+ }
+ }
+
+ if (jobStatuses.contains(ExecutionStatus.FAILED)) {
+ flowExecutionStatus = ExecutionStatus.FAILED;
+ } else if (jobStatuses.contains(ExecutionStatus.CANCELLED)) {
+ flowExecutionStatus = ExecutionStatus.CANCELLED;
+ } else if (jobStatuses.contains(ExecutionStatus.ORCHESTRATED)) {
+ flowExecutionStatus = ExecutionStatus.ORCHESTRATED;
+ } else if (jobStatuses.contains(ExecutionStatus.RUNNING)) {
+ flowExecutionStatus = ExecutionStatus.RUNNING;
+ } else if (jobStatuses.contains(ExecutionStatus.COMPLETE)) {
+ flowExecutionStatus = ExecutionStatus.COMPLETE;
Review comment:
1. this `Set`-based approach is much clearer--nice work!
2. just to check the presumption behind what I understood from the other,
iterator-traversal approach: are you certain that `ORCHESTRATED`, e.g., would
definitely be the status of the latest job (and not get picked up from a prior
job, later, say, `COMPLETE`)? or are we relying on the `KafkaJobStatusMonitor`
overwriting the status of any prior job that once had a non-final status, like
`ORCHESTRATED` with a final one like `COMPLETE`?
3. optional to convert control flow to data; e.g.:
```
List<ExecutionStatus> statusesInDescendingSalience = ImmutableList.of(F, Ca,
O, R, Co);
statusesInDescendingSalience.stream()
.filter(jobStatuses::contains).findFirst().orElse($UNKNOWN);
```
##########
File path:
gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/MysqlJobStatusRetrieverTestWithoutDagManager.java
##########
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.monitoring;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Properties;
+
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.google.common.base.Strings;
+
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.metastore.MysqlJobStatusStateStore;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.runtime.troubleshooter.MultiContextIssueRepository;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.ServiceConfigKeys;
+
+import static org.mockito.Mockito.mock;
+
+
+public class MysqlJobStatusRetrieverTestWithoutDagManager extends
JobStatusRetrieverTest {
+ private MysqlJobStatusStateStore<State> dbJobStateStore;
+ private static final String TEST_USER = "testUser";
+ private static final String TEST_PASSWORD = "testPassword";
+
+ @BeforeClass
+ @Override
+ public void setUp() throws Exception {
+ ITestMetastoreDatabase testMetastoreDatabase =
TestMetastoreDatabaseFactory.get();
+ String jdbcUrl = testMetastoreDatabase.getJdbcUrl();
+
+ ConfigBuilder configBuilder = ConfigBuilder.create();
+
configBuilder.addPrimitive(MysqlJobStatusRetriever.MYSQL_JOB_STATUS_RETRIEVER_PREFIX
+ "." + ConfigurationKeys.STATE_STORE_DB_URL_KEY, jdbcUrl);
+
configBuilder.addPrimitive(MysqlJobStatusRetriever.MYSQL_JOB_STATUS_RETRIEVER_PREFIX
+ "." + ConfigurationKeys.STATE_STORE_DB_USER_KEY, TEST_USER);
+
configBuilder.addPrimitive(MysqlJobStatusRetriever.MYSQL_JOB_STATUS_RETRIEVER_PREFIX
+ "." + ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, TEST_PASSWORD);
+
+ this.jobStatusRetriever =
+ new MysqlJobStatusRetriever(configBuilder.build(),
mock(MultiContextIssueRepository.class));
+
configBuilder.addPrimitive(ServiceConfigKeys.GOBBLIN_SERVICE_DAG_MANAGER_ENABLED_KEY,
"true");
+ this.dbJobStateStore = ((MysqlJobStatusRetriever)
this.jobStatusRetriever).getStateStore();
+ cleanUpDir();
+ }
+
+ @Test
+ public void testGetJobStatusesForFlowExecution() throws IOException {
+ super.testGetJobStatusesForFlowExecution();
+ }
+
+ @Test (dependsOnMethods = "testGetJobStatusesForFlowExecution")
+ public void testJobTiming() throws Exception {
+ super.testJobTiming();
+ }
+
+ @Test (dependsOnMethods = "testJobTiming")
+ public void testOutOfOrderJobTimingEvents() throws IOException {
+ super.testOutOfOrderJobTimingEvents();
+ }
+
+ @Test (dependsOnMethods = "testJobTiming")
+ public void testGetJobStatusesForFlowExecution1() {
+ super.testGetJobStatusesForFlowExecution1();
+ }
+
+ @Test (dependsOnMethods = "testGetJobStatusesForFlowExecution1")
+ public void testGetLatestExecutionIdsForFlow() throws Exception {
+ super.testGetLatestExecutionIdsForFlow();
+ }
+
+ @Test (dependsOnMethods = "testGetLatestExecutionIdsForFlow")
+ public void testGetFlowStatusFromJobStatuses() throws Exception {
Review comment:
nice side-by-side illustration from the `MysqlJobStatusRetrieverTest`.
##########
File path:
gobblin-service/src/test/java/org/apache/gobblin/service/monitoring/MysqlJobStatusRetrieverTestWithoutDagManager.java
##########
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service.monitoring;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Properties;
+
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.google.common.base.Strings;
+
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.metastore.MysqlJobStatusStateStore;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.runtime.troubleshooter.MultiContextIssueRepository;
+import org.apache.gobblin.service.ExecutionStatus;
+import org.apache.gobblin.service.ServiceConfigKeys;
+
+import static org.mockito.Mockito.mock;
+
+
+public class MysqlJobStatusRetrieverTestWithoutDagManager extends
JobStatusRetrieverTest {
+ private MysqlJobStatusStateStore<State> dbJobStateStore;
+ private static final String TEST_USER = "testUser";
+ private static final String TEST_PASSWORD = "testPassword";
+
+ @BeforeClass
+ @Override
+ public void setUp() throws Exception {
+ ITestMetastoreDatabase testMetastoreDatabase =
TestMetastoreDatabaseFactory.get();
+ String jdbcUrl = testMetastoreDatabase.getJdbcUrl();
+
+ ConfigBuilder configBuilder = ConfigBuilder.create();
+
configBuilder.addPrimitive(MysqlJobStatusRetriever.MYSQL_JOB_STATUS_RETRIEVER_PREFIX
+ "." + ConfigurationKeys.STATE_STORE_DB_URL_KEY, jdbcUrl);
+
configBuilder.addPrimitive(MysqlJobStatusRetriever.MYSQL_JOB_STATUS_RETRIEVER_PREFIX
+ "." + ConfigurationKeys.STATE_STORE_DB_USER_KEY, TEST_USER);
+
configBuilder.addPrimitive(MysqlJobStatusRetriever.MYSQL_JOB_STATUS_RETRIEVER_PREFIX
+ "." + ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, TEST_PASSWORD);
+
+ this.jobStatusRetriever =
+ new MysqlJobStatusRetriever(configBuilder.build(),
mock(MultiContextIssueRepository.class));
+
configBuilder.addPrimitive(ServiceConfigKeys.GOBBLIN_SERVICE_DAG_MANAGER_ENABLED_KEY,
"true");
+ this.dbJobStateStore = ((MysqlJobStatusRetriever)
this.jobStatusRetriever).getStateStore();
+ cleanUpDir();
+ }
+
+ @Test
+ public void testGetJobStatusesForFlowExecution() throws IOException {
+ super.testGetJobStatusesForFlowExecution();
+ }
+
+ @Test (dependsOnMethods = "testGetJobStatusesForFlowExecution")
+ public void testJobTiming() throws Exception {
+ super.testJobTiming();
+ }
+
+ @Test (dependsOnMethods = "testJobTiming")
+ public void testOutOfOrderJobTimingEvents() throws IOException {
+ super.testOutOfOrderJobTimingEvents();
+ }
+
+ @Test (dependsOnMethods = "testJobTiming")
+ public void testGetJobStatusesForFlowExecution1() {
+ super.testGetJobStatusesForFlowExecution1();
+ }
+
+ @Test (dependsOnMethods = "testGetJobStatusesForFlowExecution1")
+ public void testGetLatestExecutionIdsForFlow() throws Exception {
+ super.testGetLatestExecutionIdsForFlow();
+ }
+
+ @Test (dependsOnMethods = "testGetLatestExecutionIdsForFlow")
+ public void testGetFlowStatusFromJobStatuses() throws Exception {
+ long flowExecutionId = 1237L;
+
+ addJobStatusToStateStore(flowExecutionId, JobStatusRetriever.NA_KEY,
ExecutionStatus.COMPILED.name());
+ Assert.assertEquals(ExecutionStatus.$UNKNOWN,
+
jobStatusRetriever.getFlowStatusFromJobStatuses(jobStatusRetriever.dagManagerEnabled,
jobStatusRetriever.getJobStatusesForFlowExecution(FLOW_NAME, FLOW_GROUP,
flowExecutionId)));
Review comment:
nit: call statically
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 660125)
Time Spent: 1h 20m (was: 1h 10m)
> fix flow status reporting when dag manager is not enabled
> ---------------------------------------------------------
>
> Key: GOBBLIN-1552
> URL: https://issues.apache.org/jira/browse/GOBBLIN-1552
> Project: Apache Gobblin
> Issue Type: Bug
> Reporter: Arjun Singh Bora
> Priority: Major
> Time Spent: 1h 20m
> Remaining Estimate: 0h
>
> flow status is determined by looking at the flow level events. but flow level
> events are not emitted outside of dag manager. so currently flow status is
> not being determined correctly when dag manager is disabled
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
