diff --git a/jobs/jobs_api_go_sdk.go b/jobs/jobs_api_go_sdk.go
index ae1d6c32aa..2e686a6d5d 100644
--- a/jobs/jobs_api_go_sdk.go
+++ b/jobs/jobs_api_go_sdk.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/databricks/databricks-sdk-go"
 	"github.com/databricks/databricks-sdk-go/apierr"
+	"github.com/databricks/databricks-sdk-go/service/compute"
 	"github.com/databricks/databricks-sdk-go/service/jobs"
 	"github.com/databricks/terraform-provider-databricks/clusters"
 	"github.com/databricks/terraform-provider-databricks/common"
@@ -155,20 +156,39 @@ func (c controlRunStateLifecycleManagerGoSdk) OnUpdate(ctx context.Context) erro
 	return StopActiveRun(jobID, c.d.Timeout(schema.TimeoutUpdate), w, ctx)
 }
 
+// Removing AutoterminationMinutes out of ClusterSpec's ForceSendFields because the current terraform plugin sdk
+// has a limitation that it will always set unspecified values to the zero value.
+//
+// `autotermination_minutes` should never be explicitly set for jobs because automated clusters don't accept this config.
+// TODO: consider removing `autotermination_minutes` from the jobs terraform schema.
+func removeAutoterminationMinutesFromForceSendFields(clusterSpec *compute.ClusterSpec) error {
+	newForceSendFields := []string{}
+	for _, field := range clusterSpec.ForceSendFields {
+		if field != "AutoterminationMinutes" {
+			newForceSendFields = append(newForceSendFields, field)
+		}
+	}
+	clusterSpec.ForceSendFields = newForceSendFields
+	return nil
+}
+
 func prepareJobSettingsForUpdateGoSdk(d *schema.ResourceData, js JobSettingsResource) {
 	if js.NewCluster != nil {
 		clusters.ModifyRequestOnInstancePool(js.NewCluster)
 		clusters.FixInstancePoolChangeIfAny(d, &js.NewCluster)
+		removeAutoterminationMinutesFromForceSendFields(js.NewCluster)
 	}
 	for _, task := range js.Tasks {
 		if task.NewCluster != nil {
 			clusters.ModifyRequestOnInstancePool(task.NewCluster)
 			clusters.FixInstancePoolChangeIfAny(d, &task.NewCluster)
+			removeAutoterminationMinutesFromForceSendFields(task.NewCluster)
 		}
 	}
 	for i := range js.JobClusters {
 		clusters.ModifyRequestOnInstancePool(&js.JobClusters[i].NewCluster)
 		clusters.FixInstancePoolChangeIfAny(d, &js.JobClusters[i].NewCluster)
+		removeAutoterminationMinutesFromForceSendFields(&js.JobClusters[i].NewCluster)
 	}
 }
 
diff --git a/jobs/resource_job.go b/jobs/resource_job.go
index 978ac0c763..caedb525bd 100644
--- a/jobs/resource_job.go
+++ b/jobs/resource_job.go
@@ -612,14 +612,9 @@ func (JobSettingsResource) CustomizeSchema(s *common.CustomizableSchema) *common
 	s.SchemaPath("task", "for_each_task", "task", "new_cluster").RemoveField("cluster_source")
 
 	// ======= To keep consistency with the manually maintained schema, should be reverted once full migration is done. ======
-	// s.SchemaPath("task", "task_key").SetOptional()
-	// s.SchemaPath("task", "for_each_task", "task", "task_key").SetOptional()
 
 	s.SchemaPath("trigger", "table_update", "table_names").SetRequired()
 
-	// s.SchemaPath("task", "sql_task", "warehouse_id").SetOptional()
-	// s.SchemaPath("task", "for_each_task", "task", "sql_task", "warehouse_id").SetOptional()
-
 	s.SchemaPath("task", "python_wheel_task", "entry_point").SetOptional()
 	s.SchemaPath("task", "for_each_task", "task", "python_wheel_task", "entry_point").SetOptional()
 
@@ -629,10 +624,6 @@ func (JobSettingsResource) CustomizeSchema(s *common.CustomizableSchema) *common
 	s.SchemaPath("task", "sql_task", "alert", "subscriptions").SetRequired()
 	s.SchemaPath("task", "for_each_task", "task", "sql_task", "alert", "subscriptions").SetRequired()
 
-	// s.SchemaPath("health", "rules", "metric").SetOptional()
-	// s.SchemaPath("health", "rules", "op").SetOptional()
-	// s.SchemaPath("health", "rules", "value").SetOptional()
-
 	s.SchemaPath("task", "new_cluster", "cluster_id").SetOptional()
 	s.SchemaPath("task", "for_each_task", "task", "new_cluster", "cluster_id").SetOptional()
 
@@ -640,23 +631,9 @@ func (JobSettingsResource) CustomizeSchema(s *common.CustomizableSchema) *common
 	s.SchemaPath("task", "health", "rules").SetRequired()
 	s.SchemaPath("task", "for_each_task", "task", "health", "rules").SetRequired()
 
-	// s.SchemaPath("task", "health", "rules", "op").SetOptional()
-	// s.SchemaPath("task", "for_each_task", "task", "health", "rules", "op").SetOptional()
-
-	// s.SchemaPath("task", "health", "rules", "metric").SetOptional()
-	// s.SchemaPath("task", "for_each_task", "task", "health", "rules", "metric").SetOptional()
-
-	// s.SchemaPath("task", "health", "rules", "value").SetOptional()
-	// s.SchemaPath("task", "for_each_task", "task", "health", "rules", "value").SetOptional()
-
-	// s.SchemaPath("job_cluster", "job_cluster_key").SetOptional()
-	// s.SchemaPath("job_cluster", "new_cluster").SetOptional()
 	s.SchemaPath("job_cluster", "new_cluster", "cluster_id").SetOptional()
-
 	s.SchemaPath("new_cluster", "cluster_id").SetOptional()
 
-	// s.SchemaPath("git_source", "provider").SetOptional()
-
 	s.SchemaPath("library").Schema.Type = schema.TypeSet
 	s.SchemaPath("task", "library").Schema.Type = schema.TypeSet
 
@@ -1057,18 +1034,11 @@ func prepareJobSettingsForUpdate(d *schema.ResourceData, js JobSettings) {
 	}
 }
 
-func removeZeroAutoterminationMinutesFromJobSetting(d *schema.ResourceData) error {
-	if v, ok := d.GetOkExists("task.0.new_cluster.0.autotermination_minutes"); ok && v.(string) == "0" {
-		d.Set("task.0.new_cluster.0.autotermination_minutes", nil)
-	}
-	return nil
-}
-
 var jobsGoSdkSchema = common.StructToSchema(JobSettingsResource{}, nil)
 
 func ResourceJob() common.Resource {
 	getReadCtx := func(ctx context.Context, d *schema.ResourceData) context.Context {
-		var js JobSettings
+		var js JobSettingsResource
 		common.DataToStructPointer(d, jobsGoSdkSchema, &js)
 		if js.isMultiTask() {
 			return context.WithValue(ctx, common.Api, common.API_2_1)
@@ -1083,7 +1053,7 @@ func ResourceJob() common.Resource {
 			Update: schema.DefaultTimeout(clusters.DefaultProvisionTimeout),
 		},
 		CustomizeDiff: func(ctx context.Context, d *schema.ResourceDiff) error {
-			var js JobSettings
+			var js JobSettingsResource
 			common.DiffToStructPointer(d, jobsGoSdkSchema, &js)
 			alwaysRunning := d.Get("always_running").(bool)
 			if alwaysRunning && js.MaxConcurrentRuns > 1 {
@@ -1102,12 +1072,12 @@ func ResourceJob() common.Resource {
 				if task.NewCluster == nil {
 					continue
 				}
-				if err := task.NewCluster.Validate(); err != nil {
+				if err := clusters.Validate(*task.NewCluster); err != nil {
 					return fmt.Errorf("task %s invalid: %w", task.TaskKey, err)
 				}
 			}
 			if js.NewCluster != nil {
-				if err := js.NewCluster.Validate(); err != nil {
+				if err := clusters.Validate(*js.NewCluster); err != nil {
 					return fmt.Errorf("invalid job cluster: %w", err)
 				}
 			}
@@ -1145,7 +1115,6 @@ func ResourceJob() common.Resource {
 		Read: func(ctx context.Context, d *schema.ResourceData, c *common.DatabricksClient) error {
 			var js JobSettingsResource
 			common.DataToStructPointer(d, jobsGoSdkSchema, &js)
-			removeZeroAutoterminationMinutesFromJobSetting(d)
 			if js.isMultiTask() {
 				// Api 2.1
 				w, err := c.WorkspaceClient()