You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2022/05/17 13:09:39 UTC
[flink-kubernetes-operator] branch main updated: [hotfix] Do not use nulls in status for savepoint trigger and cluster info
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new f73269e [hotfix] Do not use nulls in status for savepoint trigger and cluster info
f73269e is described below
commit f73269efc17f47d0b244ffb0843f4f98487897eb
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Tue May 17 12:06:17 2022 +0200
[hotfix] Do not use nulls in status for savepoint trigger and cluster info
---
.../kubernetes/operator/crd/status/SavepointInfo.java | 8 ++++----
.../operator/observer/SavepointObserver.java | 3 ++-
.../deployment/AbstractDeploymentObserver.java | 5 +++--
.../reconciler/sessionjob/SessionJobHelper.java | 3 ++-
.../kubernetes/operator/utils/SavepointUtils.java | 4 +++-
.../observer/deployment/ApplicationObserverTest.java | 15 +++++++--------
.../observer/sessionjob/SessionJobObserverTest.java | 11 +++++++----
.../deployment/ApplicationReconcilerTest.java | 18 +++++++++++-------
.../sessionjob/FlinkSessionJobReconcilerTest.java | 19 +++++++++----------
9 files changed, 48 insertions(+), 38 deletions(-)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointInfo.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointInfo.java
index 2bd52eb..b20691d 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointInfo.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointInfo.java
@@ -38,10 +38,10 @@ public class SavepointInfo {
private Savepoint lastSavepoint;
/** Trigger id of a pending savepoint operation. */
- private String triggerId;
+ private String triggerId = "";
/** Trigger timestamp of a pending savepoint operation. */
- private Long triggerTimestamp;
+ private Long triggerTimestamp = 0L;
/** List of recent savepoints. */
private List<Savepoint> savepointHistory = new ArrayList<>();
@@ -52,8 +52,8 @@ public class SavepointInfo {
}
public void resetTrigger() {
- this.triggerId = null;
- this.triggerTimestamp = null;
+ this.triggerId = "";
+ this.triggerTimestamp = 0L;
}
public void updateLastSavepoint(Savepoint savepoint) {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
index 77c86c1..daa8142 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
@@ -32,6 +32,7 @@ import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
import org.apache.flink.kubernetes.operator.utils.StatusHelper;
+import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -101,7 +102,7 @@ public class SavepointObserver<STATUS extends CommonStatus<?>> {
*/
private Optional<String> observeTriggeredSavepointProgress(
SavepointInfo currentSavepointInfo, String jobID, Configuration deployedConfig) {
- if (currentSavepointInfo.getTriggerId() == null) {
+ if (StringUtils.isEmpty(currentSavepointInfo.getTriggerId())) {
LOG.debug("Savepoint not in progress");
return Optional.empty();
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
index af7d9f2..e275a74 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
@@ -47,6 +47,7 @@ import io.javaoperatorsdk.operator.api.reconciler.Context;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -101,7 +102,7 @@ public abstract class AbstractDeploymentObserver implements Observer<FlinkDeploy
}
private boolean observeClusterInfo(FlinkDeployment flinkApp, Configuration configuration) {
- if (flinkApp.getStatus().getClusterInfo() != null) {
+ if (!flinkApp.getStatus().getClusterInfo().isEmpty()) {
return true;
}
try {
@@ -126,7 +127,7 @@ public abstract class AbstractDeploymentObserver implements Observer<FlinkDeploy
return;
}
- flinkApp.getStatus().setClusterInfo(null);
+ flinkApp.getStatus().setClusterInfo(new HashMap<>());
logger.info(
"Observing JobManager deployment. Previous status: {}", previousJmStatus.name());
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobHelper.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobHelper.java
index 6718aee..bca3789 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobHelper.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobHelper.java
@@ -22,6 +22,7 @@ import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
import org.apache.flink.kubernetes.operator.crd.spec.FlinkSessionJobSpec;
import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
+import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
import org.slf4j.Logger;
@@ -42,7 +43,7 @@ public class SessionJobHelper {
}
public boolean savepointInProgress() {
- return sessionJob.getStatus().getJobStatus().getSavepointInfo().getTriggerId() != null;
+ return SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus());
}
public boolean shouldTriggerSavepoint() {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SavepointUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SavepointUtils.java
index cb1f6eb..8383f32 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SavepointUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SavepointUtils.java
@@ -23,13 +23,15 @@ import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
import org.apache.flink.kubernetes.operator.crd.status.SavepointInfo;
+import org.apache.commons.lang3.StringUtils;
+
import java.time.Duration;
/** Savepoint utilities. */
public class SavepointUtils {
public static boolean savepointInProgress(JobStatus jobStatus) {
- return jobStatus.getSavepointInfo().getTriggerId() != null;
+ return StringUtils.isNotEmpty(jobStatus.getSavepointInfo().getTriggerId());
}
public static boolean shouldTriggerSavepoint(JobSpec jobSpec, FlinkDeploymentStatus status) {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
index 9b800c4..78b1320 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
@@ -29,11 +29,13 @@ import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
import org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
+import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import org.junit.jupiter.api.Test;
+import static org.junit.Assert.assertFalse;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -90,7 +92,7 @@ public class ApplicationObserverTest {
JobManagerDeploymentStatus.DEPLOYED_NOT_READY,
deployment.getStatus().getJobManagerDeploymentStatus());
assertNull(deployment.getStatus().getReconciliationStatus().getLastStableSpec());
- assertNull(deployment.getStatus().getClusterInfo());
+ assertTrue(deployment.getStatus().getClusterInfo().isEmpty());
// Stable ready
observer.observe(deployment, readyContext);
@@ -185,9 +187,8 @@ public class ApplicationObserverTest {
.getSavepointInfo()
.getLastSavepoint()
.getLocation());
- assertNull(deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
- assertNull(deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerTimestamp());
+ assertFalse(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
flinkService.triggerSavepoint(
deployment.getStatus().getJobStatus().getJobId(),
@@ -196,6 +197,7 @@ public class ApplicationObserverTest {
assertEquals(
"trigger_1",
deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertTrue(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
observer.observe(deployment, readyContext);
assertEquals(
@@ -206,9 +208,7 @@ public class ApplicationObserverTest {
.getSavepointInfo()
.getLastSavepoint()
.getLocation());
- assertNull(deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
-
- assertNull(deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerTimestamp());
+ assertFalse(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
// Simulate Failed job
Tuple2<String, JobStatusMessage> jobTuple = flinkService.listJobs().get(0);
@@ -234,8 +234,7 @@ public class ApplicationObserverTest {
.getSavepointInfo()
.getLastSavepoint()
.getLocation());
- assertNull(deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerTimestamp());
- assertNull(deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertFalse(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
}
private void bringToReadyStatus(FlinkDeployment deployment) {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
index 22401fa..c5ff4eb 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.kubernetes.operator.TestingFlinkService;
import org.apache.flink.kubernetes.operator.TestingStatusHelper;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.reconciler.sessionjob.FlinkSessionJobReconciler;
+import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -145,15 +146,17 @@ public class SessionJobObserverTest {
JobStatus.RUNNING.name(), sessionJob.getStatus().getJobStatus().getState());
var savepointInfo = sessionJob.getStatus().getJobStatus().getSavepointInfo();
- Assertions.assertNull(savepointInfo.getTriggerId());
- Assertions.assertNull(savepointInfo.getTriggerTimestamp());
+ Assertions.assertFalse(
+ SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
flinkService.triggerSavepoint(jobID, savepointInfo, new Configuration());
+ Assertions.assertTrue(
+ SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
Assertions.assertEquals("trigger_0", savepointInfo.getTriggerId());
observer.observe(sessionJob, readyContext);
Assertions.assertEquals("savepoint_0", savepointInfo.getLastSavepoint().getLocation());
- Assertions.assertNull(savepointInfo.getTriggerId());
- Assertions.assertNull(savepointInfo.getTriggerTimestamp());
+ Assertions.assertFalse(
+ SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
}
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
index 354d07a..d3c3289 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
@@ -35,6 +35,7 @@ import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatu
import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
import org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.highavailability.JobResultStoreOptions;
@@ -51,6 +52,8 @@ import java.util.List;
import java.util.Optional;
import java.util.concurrent.ThreadLocalRandom;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -190,14 +193,14 @@ public class ApplicationReconcilerTest {
reconciler.reconcile(deployment, context);
List<Tuple2<String, JobStatusMessage>> runningJobs = flinkService.listJobs();
verifyAndSetRunningJobsToStatus(deployment, runningJobs);
- assertNull(deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertFalse(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
// trigger savepoint
FlinkDeployment spDeployment = ReconciliationUtils.clone(deployment);
// don't trigger if nonce is missing
reconciler.reconcile(spDeployment, context);
- assertNull(spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertFalse(SavepointUtils.savepointInProgress(spDeployment.getStatus().getJobStatus()));
// trigger when nonce is defined
spDeployment
@@ -208,6 +211,7 @@ public class ApplicationReconcilerTest {
assertEquals(
"trigger_0",
spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertTrue(SavepointUtils.savepointInProgress(spDeployment.getStatus().getJobStatus()));
// don't trigger when savepoint is in progress
reconciler.reconcile(spDeployment, context);
@@ -215,13 +219,13 @@ public class ApplicationReconcilerTest {
"trigger_0",
spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
- spDeployment.getStatus().getJobStatus().getSavepointInfo().setTriggerId(null);
+ spDeployment.getStatus().getJobStatus().getSavepointInfo().resetTrigger();
// don't trigger when nonce is the same
reconciler.reconcile(spDeployment, context);
- assertNull(spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertFalse(SavepointUtils.savepointInProgress(spDeployment.getStatus().getJobStatus()));
- spDeployment.getStatus().getJobStatus().getSavepointInfo().setTriggerId(null);
+ spDeployment.getStatus().getJobStatus().getSavepointInfo().resetTrigger();
// trigger when new nonce is defined
spDeployment
@@ -233,12 +237,12 @@ public class ApplicationReconcilerTest {
"trigger_1",
spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
- spDeployment.getStatus().getJobStatus().getSavepointInfo().setTriggerId(null);
+ spDeployment.getStatus().getJobStatus().getSavepointInfo().resetTrigger();
// don't trigger nonce is cleared
spDeployment.getSpec().getJob().setSavepointTriggerNonce(null);
reconciler.reconcile(spDeployment, context);
- assertNull(spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertFalse(SavepointUtils.savepointInProgress(spDeployment.getStatus().getJobStatus()));
}
@Test
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconcilerTest.java
index 1aea732..4aff486 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconcilerTest.java
@@ -28,6 +28,7 @@ import org.apache.flink.kubernetes.operator.crd.spec.JobState;
import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import org.junit.jupiter.api.Assertions;
@@ -38,8 +39,8 @@ import javax.annotation.Nullable;
import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
+import static org.junit.Assert.assertFalse;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
/** Tests for {@link FlinkSessionJobReconciler}. */
@@ -228,7 +229,7 @@ public class FlinkSessionJobReconcilerTest {
public void testTriggerSavepoint() throws Exception {
TestingFlinkService flinkService = new TestingFlinkService();
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
- assertNull(sessionJob.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertFalse(SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
FlinkSessionJobReconciler reconciler =
@@ -241,14 +242,14 @@ public class FlinkSessionJobReconcilerTest {
null,
flinkService.listSessionJobs());
- assertNull(sessionJob.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertFalse(SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
// trigger savepoint
var sp1SessionJob = ReconciliationUtils.clone(sessionJob);
// do not trigger savepoint if nonce is null
reconciler.reconcile(sp1SessionJob, readyContext);
- assertNull(sessionJob.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertFalse(SavepointUtils.savepointInProgress(sp1SessionJob.getStatus().getJobStatus()));
sp1SessionJob.getSpec().getJob().setSavepointTriggerNonce(2L);
sp1SessionJob
@@ -257,7 +258,7 @@ public class FlinkSessionJobReconcilerTest {
.setState(org.apache.flink.api.common.JobStatus.CREATED.name());
reconciler.reconcile(sp1SessionJob, readyContext);
// do not trigger savepoint if job is not running
- assertNull(sp1SessionJob.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertFalse(SavepointUtils.savepointInProgress(sp1SessionJob.getStatus().getJobStatus()));
sp1SessionJob
.getStatus()
@@ -265,9 +266,7 @@ public class FlinkSessionJobReconcilerTest {
.setState(org.apache.flink.api.common.JobStatus.RUNNING.name());
reconciler.reconcile(sp1SessionJob, readyContext);
- assertEquals(
- "trigger_0",
- sp1SessionJob.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertTrue(SavepointUtils.savepointInProgress(sp1SessionJob.getStatus().getJobStatus()));
// the last reconcile nonce updated
assertEquals(
@@ -338,7 +337,7 @@ public class FlinkSessionJobReconcilerTest {
// don't trigger when nonce is the same
sp1SessionJob.getSpec().getJob().setSavepointTriggerNonce(2L);
reconciler.reconcile(sp1SessionJob, readyContext);
- assertNull(sp1SessionJob.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertFalse(SavepointUtils.savepointInProgress(sp1SessionJob.getStatus().getJobStatus()));
// trigger when new nonce is defined
sp1SessionJob.getSpec().getJob().setSavepointTriggerNonce(3L);
@@ -352,7 +351,7 @@ public class FlinkSessionJobReconcilerTest {
// don't trigger when nonce is cleared
sp1SessionJob.getSpec().getJob().setSavepointTriggerNonce(null);
reconciler.reconcile(sp1SessionJob, readyContext);
- assertNull(sp1SessionJob.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertFalse(SavepointUtils.savepointInProgress(sp1SessionJob.getStatus().getJobStatus()));
}
private TestingFlinkService.SubmittedJobInfo verifyAndReturnTheSubmittedJob(