direct: fix drift detection for jobs with >100 tasks (#4675)

denik · claude · web-flow · commit 74899eb37285 · 2026-03-06T16:15:37.000+01:00
## Changes
Use jobs.Get() instead of jobs.GetById() to properly paginate tasks.

## Why
Fixes plan for jobs with more than 100 tasks.

## Tests
New pydabs configs for invariant tests for 10 and 1000 jobs. The 1k
config failed no_drift test because remote state only had 100 tasks and
the reset were considered 'update'.

Testserver for jobs is extended to read all jobs.

---------

Co-authored-by: Claude Sonnet 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,5 +1,11 @@
 # Version changelog
 
+## Release v0.293.0 (2026-03-06)
+
+### Bundles
+* direct: fix drift detection for jobs with >100 tasks by paginating all tasks when reading job state ([#4675](https://github.com/databricks/cli/pull/4675))
+
+
 ## Release v0.292.0 (2026-03-05)
 
 ### Bundles
diff --git a/acceptance/bundle/invariant/configs/job_pydabs_1000_tasks.yml.tmpl b/acceptance/bundle/invariant/configs/job_pydabs_1000_tasks.yml.tmpl
@@ -0,0 +1,7 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+python:
+  venv_path: .venv
+  resources:
+    - "job_pydabs_1000_tasks:load_resources"
diff --git a/acceptance/bundle/invariant/configs/job_pydabs_1000_tasks.yml.tmpl-init.sh b/acceptance/bundle/invariant/configs/job_pydabs_1000_tasks.yml.tmpl-init.sh
@@ -0,0 +1,4 @@
+#!/bin/bash
+
+uv venv --quiet
+uv pip install --quiet "$DATABRICKS_BUNDLES_WHEEL"
diff --git a/acceptance/bundle/invariant/configs/job_pydabs_10_tasks.yml.tmpl b/acceptance/bundle/invariant/configs/job_pydabs_10_tasks.yml.tmpl
@@ -0,0 +1,7 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+python:
+  venv_path: .venv
+  resources:
+    - "job_pydabs_10_tasks:load_resources"
diff --git a/acceptance/bundle/invariant/configs/job_pydabs_10_tasks.yml.tmpl-init.sh b/acceptance/bundle/invariant/configs/job_pydabs_10_tasks.yml.tmpl-init.sh
@@ -0,0 +1,4 @@
+#!/bin/bash
+
+uv venv --quiet
+uv pip install --quiet "$DATABRICKS_BUNDLES_WHEEL"
diff --git a/acceptance/bundle/invariant/data/job_pydabs_1000_tasks.py b/acceptance/bundle/invariant/data/job_pydabs_1000_tasks.py
@@ -0,0 +1,38 @@
+import os
+
+from databricks.bundles.core import Resources
+
+
+def load_resources() -> Resources:
+    unique_name = os.environ["UNIQUE_NAME"]
+    spark_version = os.environ.get("DEFAULT_SPARK_VERSION", "13.3.x-scala2.12")
+    node_type_id = os.environ.get("NODE_TYPE_ID", "i3.xlarge")
+
+    resources = Resources()
+    resources.add_job(
+        resource_name="foo",
+        job={
+            "name": f"test-job-{unique_name}",
+            "tasks": [
+                {
+                    "task_key": f"task_{i:04d}",
+                    "notebook_task": {
+                        "notebook_path": "/Shared/notebook",
+                    },
+                    "job_cluster_key": "main_cluster",
+                }
+                for i in range(1000)
+            ],
+            "job_clusters": [
+                {
+                    "job_cluster_key": "main_cluster",
+                    "new_cluster": {
+                        "spark_version": spark_version,
+                        "node_type_id": node_type_id,
+                        "num_workers": 1,
+                    },
+                }
+            ],
+        },
+    )
+    return resources
diff --git a/acceptance/bundle/invariant/data/job_pydabs_10_tasks.py b/acceptance/bundle/invariant/data/job_pydabs_10_tasks.py
@@ -0,0 +1,38 @@
+import os
+
+from databricks.bundles.core import Resources
+
+
+def load_resources() -> Resources:
+    unique_name = os.environ["UNIQUE_NAME"]
+    spark_version = os.environ.get("DEFAULT_SPARK_VERSION", "13.3.x-scala2.12")
+    node_type_id = os.environ.get("NODE_TYPE_ID", "i3.xlarge")
+
+    resources = Resources()
+    resources.add_job(
+        resource_name="foo",
+        job={
+            "name": f"test-job-{unique_name}",
+            "tasks": [
+                {
+                    "task_key": f"task_{i:02d}",
+                    "notebook_task": {
+                        "notebook_path": "/Shared/notebook",
+                    },
+                    "job_cluster_key": "main_cluster",
+                }
+                for i in range(10)
+            ],
+            "job_clusters": [
+                {
+                    "job_cluster_key": "main_cluster",
+                    "new_cluster": {
+                        "spark_version": spark_version,
+                        "node_type_id": node_type_id,
+                        "num_workers": 1,
+                    },
+                }
+            ],
+        },
+    )
+    return resources
diff --git a/acceptance/bundle/invariant/migrate/out.test.toml b/acceptance/bundle/invariant/migrate/out.test.toml
diff --git a/acceptance/bundle/invariant/no_drift/out.test.toml b/acceptance/bundle/invariant/no_drift/out.test.toml
diff --git a/acceptance/bundle/invariant/test.toml b/acceptance/bundle/invariant/test.toml
@@ -1,9 +1,11 @@
 Local = true
 Cloud = true
 RequiresUnityCatalog = true
+Timeout = '10m'
 
 Ignore = [
     ".databricks",
+    ".venv",
     "databricks.yml",
     "plan.json",
     "*.py",
@@ -28,6 +30,8 @@ EnvMatrix.INPUT_CONFIG = [
     "experiment.yml.tmpl",
     "external_location.yml.tmpl",
     "job.yml.tmpl",
+    "job_pydabs_10_tasks.yml.tmpl",
+    "job_pydabs_1000_tasks.yml.tmpl",
     "job_with_task.yml.tmpl",
     "model.yml.tmpl",
     "model_serving_endpoint.yml.tmpl",
diff --git a/bundle/direct/dresources/job.go b/bundle/direct/dresources/job.go
@@ -70,7 +70,13 @@ func (r *ResourceJob) DoRead(ctx context.Context, id string) (*JobRemote, error)
 	if err != nil {
 		return nil, err
 	}
-	job, err := r.client.Jobs.GetByJobId(ctx, idInt)
+	// GetByJobId only fetches the first page (100 tasks). Jobs.Get handles
+	// pagination and returns the complete job with all tasks merged.
+	job, err := r.client.Jobs.Get(ctx, jobs.GetJobRequest{
+		JobId:           idInt,
+		PageToken:       "",
+		ForceSendFields: nil,
+	})
 	if err != nil {
 		return nil, err
 	}
diff --git a/libs/testserver/jobs.go b/libs/testserver/jobs.go
@@ -146,6 +146,10 @@ func jobFixUps(jobSettings *jobs.JobSettings) {
 	}
 }
 
+// jobsGetTasksPageSize matches the real Databricks API limit of 100 tasks per jobs.get response.
+// https://docs.databricks.com/api/workspace/jobs/get
+const jobsGetTasksPageSize = 100
+
 func (s *FakeWorkspace) JobsGet(req Request) Response {
 	id := req.URL.Query().Get("job_id")
 	jobIdInt, err := strconv.ParseInt(id, 10, 64)
@@ -164,6 +168,42 @@ func (s *FakeWorkspace) JobsGet(req Request) Response {
 	}
 
 	job = setSourceIfNotSet(job)
+
+	// Paginate tasks to match real API behavior: max 100 tasks per response.
+	// Use page_token (the offset as a string) to fetch subsequent pages.
+	if job.Settings != nil && len(job.Settings.Tasks) > jobsGetTasksPageSize {
+		offset := 0
+		if pageToken := req.URL.Query().Get("page_token"); pageToken != "" {
+			offset, err = strconv.Atoi(pageToken)
+			if err != nil {
+				return Response{
+					StatusCode: 400,
+					Body:       fmt.Sprintf("Failed to parse page_token: %s", err),
+				}
+			}
+		}
+
+		settingsCopy := *job.Settings
+		job.Settings = &settingsCopy
+
+		tasks := settingsCopy.Tasks
+		end := min(offset+jobsGetTasksPageSize, len(tasks))
+		job.Settings.Tasks = tasks[offset:end]
+
+		if end < len(tasks) {
+			job.HasMore = true
+			job.NextPageToken = strconv.Itoa(end)
+		}
+
+		// On subsequent pages the real API returns empty arrays for properties
+		// that were fully included on the first page.
+		if offset > 0 {
+			job.Settings.JobClusters = nil
+			job.Settings.Environments = nil
+			job.Settings.Parameters = nil
+		}
+	}
+
 	return Response{Body: job}
 }
 
