Preview PipelineJob submit() and rerun() ignore the job's project, location, credentials and create_request_timeout #7239

The preview _PipelineJob.submit() and _PipelineJob.rerun() (google/cloud/aiplatform/preview/pipelinejob/pipeline_jobs.py) ignore the project, location and credentials the job was created with. They build the request parent and the v1beta1 client from aiplatform.init() instead, and submit() also drops create_request_timeout. The GA PipelineJob.submit() uses the job's own parent, client and credentials and passes the timeout, so the two disagree.

Since rerun() and default_runtime only exist on the preview class, anyone using them with a job in a different project or region than aiplatform.init() creates the job in the wrong place, or gets a permission error with the wrong credentials.

Environment details

• OS type and version: Linux
• Python version: 3.11.16
• google-cloud-aiplatform version: main (9b9f2b3), reported version 2.6.0

Steps to reproduce

1. aiplatform.init(project="global-project", location="us-central1", credentials=global_creds)
2. Create a job with its own project="job-project", location="europe-west4" and credentials=job_creds
3. Call submit(create_request_timeout=30) on the GA PipelineJob and on the preview _PipelineJob, and rerun(...) on the preview one

Code example

Only the two RPCs are replaced by recorders that print where each request goes (the GA pipeline root bucket check is stubbed so nothing leaves the machine):

from google.auth.credentials import AnonymousCredentials
from google.cloud import aiplatform
from google.cloud.aiplatform.preview.pipelinejob import pipeline_jobs as preview_pipeline_jobs

global_creds, job_creds = AnonymousCredentials(), AnonymousCredentials()
aiplatform.init(project="global-project", location="us-central1",
                staging_bucket="gs://global-bucket", credentials=global_creds)
kwargs = dict(display_name="d", template_path="spec.json", pipeline_root="gs://job-bucket/root",
              project="job-project", location="europe-west4", credentials=job_creds)

aiplatform.PipelineJob(job_id="ga-job", **kwargs).submit(create_request_timeout=30)
preview_pipeline_jobs._PipelineJob(job_id="preview-job", **kwargs).submit(create_request_timeout=30)

Output

GA PipelineJob.submit(create_request_timeout=30):
  create: endpoint=europe-west4-aiplatform.googleapis.com:443 parent=projects/job-project/locations/europe-west4 timeout=30 credentials=job
preview _PipelineJob.submit(create_request_timeout=30):
  create: endpoint=us-central1-aiplatform.googleapis.com:443 parent=projects/global-project/locations/us-central1 timeout=None credentials=global
preview _PipelineJob.rerun(...):
  get:    endpoint=us-central1-aiplatform.googleapis.com:443 credentials=global
  create: endpoint=us-central1-aiplatform.googleapis.com:443 parent=projects/global-project/locations/us-central1 timeout=None credentials=global

Expected: the preview class sends both requests to the job's region and project with the job's credentials, and submit() passes create_request_timeout, like the GA class.

I have a small fix with tests ready and will open a PR that references this issue.