[BEAM-4665] Allow joining a running dataflow pipeline without throwing

sdks/go/pkg/beam/runners/dataflow/dataflowlib/execute.go

@@ -30,6 +30,7 @@ import (
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/runners/universal/runnerlib"
 	"github.com/golang/protobuf/proto"
 	df "google.golang.org/api/dataflow/v1b3"
+	"google.golang.org/api/googleapi"
 )
 
 // Execute submits a pipeline as a Dataflow job.
@@ -99,10 +100,16 @@ func Execute(ctx context.Context, raw *pipepb.Pipeline, opts *JobOptions, worker
 		return presult, err
 	}
 	upd, err := Submit(ctx, client, opts.Project, opts.Region, job)
+	// When in async mode, if we get a 409 because we've already submitted an actively running job with the same name
+	// just return the existing job as a convenience
+	if gErr, ok := err.(*googleapi.Error); async && ok && gErr.Code == 409 {
+		log.Info(ctx, "Unable to submit job because job with same name is already actively running. Querying Dataflow for existing job")
+		upd, err = GetRunningJobByName(client, opts.Project, opts.Region, job.Name)
+	}
 	if err != nil {
 		return presult, err
 	}
-	log.Infof(ctx, "Submitted job: %v", upd.Id)
+
 	if endpoint == "" {
 		log.Infof(ctx, "Console: https://console.cloud.google.com/dataflow/jobs/%v/%v?project=%v", opts.Region, upd.Id, opts.Project)
 	}

sdks/go/pkg/beam/runners/dataflow/dataflowlib/job.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime"
+
 	// Importing to get the side effect of the remote execution hook. See init().
 	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/harness/init"
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/pipelinex"
@@ -223,7 +224,11 @@ func Translate(ctx context.Context, p *pipepb.Pipeline, opts *JobOptions, worker
 
 // Submit submits a prepared job to Cloud Dataflow.
 func Submit(ctx context.Context, client *df.Service, project, region string, job *df.Job) (*df.Job, error) {
-	return client.Projects.Locations.Jobs.Create(project, region, job).Do()
+	upd, err := client.Projects.Locations.Jobs.Create(project, region, job).Do()
+	if err == nil {
+		log.Infof(ctx, "Submitted job: %v", upd.Id)
+	}
+	return upd, err
 }
 
 // WaitForCompletion monitors the given job until completion. It logs any messages
@@ -276,6 +281,28 @@ func NewClient(ctx context.Context, endpoint string) (*df.Service, error) {
 	return client, nil
 }
 
+// GetRunningJobByName gets a Dataflow job running by its name and returns an
+// error if none match.
+func GetRunningJobByName(client *df.Service, project, region string, name string) (*df.Job, error) {
+	jobsListCall := client.Projects.Locations.Jobs.List(project, region)
+	jobsListCall.Filter("ACTIVE")
+	jobsResponse, err := jobsListCall.Do()
+	for len(jobsResponse.Jobs) > 0 {
+		if err != nil {
+			return nil, err
+		}
+		for _, job := range jobsResponse.Jobs {
+			if job.Name == name {
+				return job, nil
+			}
+		}
+
+		jobsListCall.PageToken(jobsResponse.NextPageToken)
+		jobsResponse, err = jobsListCall.Do()
+	}
+	return nil, errors.New(fmt.Sprintf("Unable to find running job with name %s", name))
+}
+
 // GetMetrics returns a collection of metrics describing the progress of a
 // job by making a call to Cloud Monitoring service.
 func GetMetrics(ctx context.Context, client *df.Service, project, region, jobID string) (*df.JobMetrics, error) {