diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/JobsScheduler.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/JobsScheduler.java
index 43478a9c1..a46aa18d1 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/JobsScheduler.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/JobsScheduler.java
@@ -265,6 +265,19 @@ public static void main(String[] args) {
                   }
                   log.info("Shutdown hook completed.");
                 }));
+    log.info(
+        "OH_SCHED_START job_type={} numParallelJobs={} tasksWaitHours={} numJobsSubmitter={} "
+            + "numJobsPoller={} taskPollIntervalMs={} taskQueuedTimeoutMs={} taskTimeoutMs={} "
+            + "multiOperationMode={}",
+        operationType.getValue(),
+        getNumParallelJobs(cmdLine),
+        getTasksWaitHours(cmdLine),
+        getNumJobsSubmitter(cmdLine),
+        getNumJobsPoller(cmdLine),
+        getTaskPollIntervalMs(cmdLine),
+        getTaskQueuedTimeoutMs(cmdLine),
+        getTaskTimeoutMs(cmdLine),
+        isMultiOperationMode(cmdLine));
     app.run(
         operationType,
         operationTaskCls.toString(),
diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/OperationTask.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/OperationTask.java
index 155e0b18b..fcd31fbb5 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/OperationTask.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/OperationTask.java
@@ -170,16 +170,24 @@ private Optional<JobState> submitJob(Attributes typeAttributes) {
 
   protected Optional<JobState> pollJobStatus(Attributes typeAttributes) {
     long startTime = System.currentTimeMillis();
+    Optional<JobState> lastObservedState = Optional.empty();
     try {
       Optional<JobState> jobState;
       do {
         jobState = jobsClient.getState(jobId);
+        lastObservedState = jobState;
         long elapsedTime = System.currentTimeMillis() - startTime;
         // Exit status check if a job is queued for more than queuedTimeoutMs.
         if (elapsedTime > queuedTimeoutMs) {
           if (jobState.isPresent() && jobState.get().equals(JobState.QUEUED)) {
+            String executionId =
+                jobsClient.getJob(jobId).map(JobResponseBody::getExecutionId).orElse(null);
             log.info(
-                "Exiting status check for {} for {} due to queued timeout", getType(), metadata);
+                "Exiting status check for {} for {} due to queued timeout lastObservedState={} executionId={}",
+                getType(),
+                metadata,
+                lastObservedState.orElse(null),
+                executionId);
             break;
           }
         }
diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/OperationTasksBuilder.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/OperationTasksBuilder.java
index 62f007302..ced1614d7 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/OperationTasksBuilder.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/OperationTasksBuilder.java
@@ -441,6 +441,10 @@ private void buildOperationTaskListInParallelInternal(
         .doAfterTerminate(
             () -> {
               operationTaskManager.updateDataGenerationCompletion();
+              log.info(
+                  "OH_SCHED_ELIGIBLE job_type={} count={}",
+                  jobType,
+                  operationTaskManager.getTotalDataCount());
               log.info(
                   "The metadata fetched count: {} for the job type: {}",
                   operationTaskManager.getTotalDataCount(),
@@ -485,6 +489,10 @@ private void buildDataLayoutOperationTaskListInParallel(
         .doAfterTerminate(
             () -> {
               operationTaskManager.updateDataGenerationCompletion();
+              log.info(
+                  "OH_SCHED_ELIGIBLE job_type={} count={}",
+                  jobType,
+                  operationTaskManager.getTotalDataCount());
               log.info(
                   "The metadata fetched count: {} for the job type: {}",
                   operationTaskManager.getTotalDataCount(),
@@ -516,6 +524,10 @@ private void buildDatabaseLevelOperationTasksInParallel(
         .doAfterTerminate(
             () -> {
               operationTaskManager.updateDataGenerationCompletion();
+              log.info(
+                  "OH_SCHED_ELIGIBLE job_type={} count={}",
+                  jobType,
+                  operationTaskManager.getTotalDataCount());
               log.info(
                   "The metadata fetched count: {} for the job type: {}",
                   operationTaskManager.getTotalDataCount(),