address review comments

Author: xiaoweim

pkg/controller/direct/dataproc/job_controller.go

@@ -79,19 +79,6 @@ func (m *dataprocJobModel) AdapterForObject(ctx context.Context, reader client.R
 		return nil, fmt.Errorf("error converting to %T: %w", obj, err)
 	}
 
-	projectRef, err := refs.ResolveProject(ctx, reader, obj.GetNamespace(), obj.Spec.ProjectRef)
-	if err != nil {
-		return nil, err
-	}
-	projectID := projectRef.ProjectID
-	if projectID == "" {
-		return nil, fmt.Errorf("cannot resolve project")
-	}
-	location := obj.Spec.Location
-	if location == "" {
-		return nil, fmt.Errorf("region is required")
-	}
-
 	// The job ID is part of the job placement config in the proto, not the top-level resource ID.
 	// The job resource itself uses a UUID generated by the service (`reference.jobId`).
 	// We construct an internal ID representation that includes the project and location.
@@ -101,44 +88,6 @@ func (m *dataprocJobModel) AdapterForObject(ctx context.Context, reader client.R
 		return nil, fmt.Errorf("failed to create new Job identity, error: %w", err)
 	}
 
-	mapCtx := &direct.MapContext{}
-	desiredProto := DataprocJobSpec_ToProto(mapCtx, &obj.Spec)
-	if mapCtx.Err() != nil {
-		return nil, mapCtx.Err()
-	}
-	// We need the full Job message, not just the spec part.
-	desired := &pb.Job{
-		Placement:              desiredProto.Placement,
-		JobUuid:                desiredProto.JobUuid,
-		Scheduling:             desiredProto.Scheduling,
-		YarnApplications:       desiredProto.YarnApplications,
-		DriverSchedulingConfig: desiredProto.DriverSchedulingConfig,
-		Done:                   desiredProto.Done,
-		Labels:                 desiredProto.Labels,
-	}
-	// Populate the job type specific field
-	if obj.Spec.HadoopJob != nil {
-		desired.TypeJob = &pb.Job_HadoopJob{HadoopJob: desiredProto.GetHadoopJob()}
-	} else if obj.Spec.SparkJob != nil {
-		desired.TypeJob = &pb.Job_SparkJob{SparkJob: desiredProto.GetSparkJob()}
-	} else if obj.Spec.PysparkJob != nil {
-		desired.TypeJob = &pb.Job_PysparkJob{PysparkJob: desiredProto.GetPysparkJob()}
-	} else if obj.Spec.HiveJob != nil {
-		desired.TypeJob = &pb.Job_HiveJob{HiveJob: desiredProto.GetHiveJob()}
-	} else if obj.Spec.PigJob != nil {
-		desired.TypeJob = &pb.Job_PigJob{PigJob: desiredProto.GetPigJob()}
-	} else if obj.Spec.SparkRJob != nil {
-		desired.TypeJob = &pb.Job_SparkRJob{SparkRJob: desiredProto.GetSparkRJob()}
-	} else if obj.Spec.SparkSQLJob != nil {
-		desired.TypeJob = &pb.Job_SparkSqlJob{SparkSqlJob: desiredProto.GetSparkSqlJob()}
-	} else if obj.Spec.PrestoJob != nil {
-		desired.TypeJob = &pb.Job_PrestoJob{PrestoJob: desiredProto.GetPrestoJob()}
-	} else if obj.Spec.TrinoJob != nil {
-		desired.TypeJob = &pb.Job_TrinoJob{TrinoJob: desiredProto.GetTrinoJob()}
-	} else {
-		return nil, fmt.Errorf("unknown job type specified")
-	}
-
 	gcpClient, err := m.client(ctx)
 	if err != nil {
 		return nil, err
@@ -147,7 +96,7 @@ func (m *dataprocJobModel) AdapterForObject(ctx context.Context, reader client.R
 	return &dataprocJobAdapter{
 		gcpClient:   gcpClient,
 		id:          jobID,
-		desired:     desired,
+		desired:     obj,
 		generatedId: direct.LazyPtr(jobID.ID()),
 	}, nil
 }
@@ -162,7 +111,7 @@ var _ directbase.Adapter = &dataprocJobAdapter{}
 type dataprocJobAdapter struct {
 	gcpClient   *dataproc.JobControllerClient
 	id          *krm.JobIdentity // Contains ProjectID, Region, and potentially JobID (after creation or from URL)
-	desired     *pb.Job          // Desired state constructed from KRM spec
+	desired     *krm.DataprocJob // Desired state
 	actual      *pb.Job          // Actual state fetched from GCP
 	generatedId *string
 }
@@ -212,30 +161,69 @@ func (a *dataprocJobAdapter) Find(ctx context.Context) (bool, error) {
 func (a *dataprocJobAdapter) Create(ctx context.Context, createOp *directbase.CreateOperation) error {
 	klog.V(2).Infof("creating dataproc job in project %s region %s", a.id.Parent().ProjectID, a.id.Parent().Location)
 
+	mapCtx := &direct.MapContext{}
+	desiredProto := DataprocJobSpec_ToProto(mapCtx, &a.desired.Spec)
+	if mapCtx.Err() != nil {
+		return mapCtx.Err()
+	}
+	// We need the full Job message, not just the spec part.
+	desired := &pb.Job{
+		Placement:              desiredProto.Placement,
+		JobUuid:                desiredProto.JobUuid,
+		Scheduling:             desiredProto.Scheduling,
+		YarnApplications:       desiredProto.YarnApplications,
+		DriverSchedulingConfig: desiredProto.DriverSchedulingConfig,
+		Done:                   desiredProto.Done,
+		Labels:                 desiredProto.Labels,
+	}
+	// Populate the job type specific field
+	if a.desired.Spec.HadoopJob != nil {
+		desired.TypeJob = &pb.Job_HadoopJob{HadoopJob: desiredProto.GetHadoopJob()}
+	} else if a.desired.Spec.SparkJob != nil {
+		desired.TypeJob = &pb.Job_SparkJob{SparkJob: desiredProto.GetSparkJob()}
+	} else if a.desired.Spec.PysparkJob != nil {
+		desired.TypeJob = &pb.Job_PysparkJob{PysparkJob: desiredProto.GetPysparkJob()}
+	} else if a.desired.Spec.HiveJob != nil {
+		desired.TypeJob = &pb.Job_HiveJob{HiveJob: desiredProto.GetHiveJob()}
+	} else if a.desired.Spec.PigJob != nil {
+		desired.TypeJob = &pb.Job_PigJob{PigJob: desiredProto.GetPigJob()}
+	} else if a.desired.Spec.SparkRJob != nil {
+		desired.TypeJob = &pb.Job_SparkRJob{SparkRJob: desiredProto.GetSparkRJob()}
+	} else if a.desired.Spec.SparkSQLJob != nil {
+		desired.TypeJob = &pb.Job_SparkSqlJob{SparkSqlJob: desiredProto.GetSparkSqlJob()}
+	} else if a.desired.Spec.PrestoJob != nil {
+		desired.TypeJob = &pb.Job_PrestoJob{PrestoJob: desiredProto.GetPrestoJob()}
+	} else if a.desired.Spec.TrinoJob != nil {
+		desired.TypeJob = &pb.Job_TrinoJob{TrinoJob: desiredProto.GetTrinoJob()}
+	} else {
+		return fmt.Errorf("unknown job type specified")
+	}
+
 	// Set the user-specified ID if provided in the spec.
-	if a.desired.GetPlacement() != nil && a.desired.GetPlacement().GetClusterUuid() != "" {
+	if desired.GetPlacement() != nil && desired.GetPlacement().GetClusterUuid() != "" {
 		// Placement.JobUuid is deprecated. Prefer setting reference.job_id
 	} else {
 		// If no specific job ID is in the spec, use the KRM resource name/resourceID
 		// as the reference ID for submission. This makes jobs findable later if the
 		// service-generated UUID is lost.
-		if a.desired.Reference == nil {
-			a.desired.Reference = &pb.JobReference{}
+		if desired.Reference == nil {
+			desired.Reference = &pb.JobReference{}
 		}
 	}
-	a.desired.Reference.JobId = a.id.ID()
-	if a.desired.Labels == nil {
-		a.desired.Labels = make(map[string]string)
+
+	desired.Reference.JobId = a.id.ID()
+	if desired.Labels == nil {
+		desired.Labels = make(map[string]string)
 	}
-	a.desired.Labels["managed-by-cnrm"] = "true"
+	desired.Labels["managed-by-cnrm"] = "true"
 
 	// Ensure placement is non-nil if it wasn't set, as it's often needed implicitly
 	// even if empty (e.g., for cluster selectors). The mapping function should handle this.
 
 	req := &pb.SubmitJobRequest{
 		ProjectId: a.id.Parent().ProjectID,
 		Region:    a.id.Parent().Location,
-		Job:       a.desired,
+		Job:       desired,
 	}
 
 	submittedJob, err := a.gcpClient.SubmitJob(ctx, req)
@@ -247,7 +235,6 @@ func (a *dataprocJobAdapter) Create(ctx context.Context, createOp *directbase.Cr
 	a.actual = submittedJob // Store the immediate result as 'actual' state
 
 	// Map the *returned* job state to the KRM status
-	mapCtx := &direct.MapContext{}
 	status := &krm.DataprocJobStatus{}
 	status.ObservedState = DataprocJobObservedState_FromProto(mapCtx, submittedJob)
 	if mapCtx.Err() != nil {
@@ -277,8 +264,46 @@ func (a *dataprocJobAdapter) Update(ctx context.Context, updateOp *directbase.Up
 		return fmt.Errorf("actual state is nil during update for job %q", a.id)
 	}
 
+	mapCtx := &direct.MapContext{}
+	desiredProto := DataprocJobSpec_ToProto(mapCtx, &a.desired.Spec)
+	if mapCtx.Err() != nil {
+		return mapCtx.Err()
+	}
+	// We need the full Job message, not just the spec part.
+	desired := &pb.Job{
+		Placement:              desiredProto.Placement,
+		JobUuid:                desiredProto.JobUuid,
+		Scheduling:             desiredProto.Scheduling,
+		YarnApplications:       desiredProto.YarnApplications,
+		DriverSchedulingConfig: desiredProto.DriverSchedulingConfig,
+		Done:                   desiredProto.Done,
+		Labels:                 desiredProto.Labels,
+	}
+	// Populate the job type specific field
+	if a.desired.Spec.HadoopJob != nil {
+		desired.TypeJob = &pb.Job_HadoopJob{HadoopJob: desiredProto.GetHadoopJob()}
+	} else if a.desired.Spec.SparkJob != nil {
+		desired.TypeJob = &pb.Job_SparkJob{SparkJob: desiredProto.GetSparkJob()}
+	} else if a.desired.Spec.PysparkJob != nil {
+		desired.TypeJob = &pb.Job_PysparkJob{PysparkJob: desiredProto.GetPysparkJob()}
+	} else if a.desired.Spec.HiveJob != nil {
+		desired.TypeJob = &pb.Job_HiveJob{HiveJob: desiredProto.GetHiveJob()}
+	} else if a.desired.Spec.PigJob != nil {
+		desired.TypeJob = &pb.Job_PigJob{PigJob: desiredProto.GetPigJob()}
+	} else if a.desired.Spec.SparkRJob != nil {
+		desired.TypeJob = &pb.Job_SparkRJob{SparkRJob: desiredProto.GetSparkRJob()}
+	} else if a.desired.Spec.SparkSQLJob != nil {
+		desired.TypeJob = &pb.Job_SparkSqlJob{SparkSqlJob: desiredProto.GetSparkSqlJob()}
+	} else if a.desired.Spec.PrestoJob != nil {
+		desired.TypeJob = &pb.Job_PrestoJob{PrestoJob: desiredProto.GetPrestoJob()}
+	} else if a.desired.Spec.TrinoJob != nil {
+		desired.TypeJob = &pb.Job_TrinoJob{TrinoJob: desiredProto.GetTrinoJob()}
+	} else {
+		return fmt.Errorf("unknown job type specified")
+	}
+
 	// Create copies for comparison, normalizing potential output-only fields
-	desiredComparable := proto.Clone(a.desired).(*pb.Job)
+	desiredComparable := proto.Clone(desired).(*pb.Job)
 	actualComparable := proto.Clone(a.actual).(*pb.Job)
 
 	// Clear fields that are not part of the spec or are output-only before comparison