diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
index e8cf1a40b36..ab4f73f9f7b 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
@@ -596,6 +596,8 @@ private void submitJob(DagNode<JobExecutionPlan> dagNode) {
         // blocks (by calling Future#get()) until the submission is completed.
         Future addSpecFuture = producer.addSpec(jobSpec);
         dagNode.getValue().setJobFuture(Optional.of(addSpecFuture));
+        //Persist the dag
+        this.dagStateStore.writeCheckpoint(this.dags.get(DagManagerUtils.generateDagId(dagNode)));
 
         if (this.metricContext != null) {
           getRunningJobsCounter(dagNode).inc();