Repository: googlecloudplatform/dataflowtemplates
Issue: [Bug]: Datastream to BigQuery MERGE not running in correct project

### Related Template(s)

Datastream to BigQuery

### What happened?

I'm running this template using a different BigQuery project than the Datastream/GCS/Dataflow resources. I'm supplying it as the `--outputProjectId` value and data is flowing into my temporary table, just the MERGE fails:

```
2022-09-09T20:40:23.171ZMerge Job Failed: Exception: com.google.cloud.bigquery.BigQueryException: Access Denied: Project <dataflowproject>: User does not have bigquery.jobs.create permission in project <dataflowproject>.
```

I'm not super familiar with the code, but should this [line](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/4a0664a0d7559ba81353f48dc1cd7e2e928a3fd1/v2/datastream-to-bigquery/src/main/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQuery.java#L422) use the `options.getOutputProjectId()` value instead of the variable?

### Beam Version

Newer than 2.35.0

### Relevant log output

_No response_

Changed files:
v2/common/src/main/java/com/google/cloud/teleport/v2/cdc/merge/BigQueryMerger.java
v2/common/src/main/java/com/google/cloud/teleport/v2/cdc/merge/MergeConfiguration.java
v2/datastream-to-bigquery/src/main/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQuery.java

diff --git a/v2/common/src/main/java/com/google/cloud/teleport/v2/cdc/merge/BigQueryMerger.java b/v2/common/src/main/java/com/google/cloud/teleport/v2/cdc/merge/BigQueryMerger.java
index 083478a5..ad6e0781 100644
--- a/v2/common/src/main/java/com/google/cloud/teleport/v2/cdc/merge/BigQueryMerger.java
+++ b/v2/common/src/main/java/com/google/cloud/teleport/v2/cdc/merge/BigQueryMerger.java
@@ -171,7 +171,11 @@ public class BigQueryMerger extends PTransform<PCollection<MergeInfo>, PCollecti
     @Setup
     public void setUp() {
       if (bigQueryClient == null) {
-        bigQueryClient = BigQueryOptions.getDefaultInstance().getService();
+        BigQueryOptions.Builder optionsBuilder = BigQueryOptions.newBuilder();
+        if (mergeConfiguration.projectId() != null && !mergeConfiguration.projectId().isEmpty()) {
+          optionsBuilder = optionsBuilder.setProjectId(mergeConfiguration.projectId());
+        }
+        bigQueryClient = optionsBuilder.build().getService();
       }
     }
 
diff --git a/v2/common/src/main/java/com/google/cloud/teleport/v2/cdc/merge/MergeConfiguration.java b/v2/common/src/main/java/com/google/cloud/teleport/v2/cdc/merge/MergeConfiguration.java
index f796823a..14914ab3 100644
--- a/v2/common/src/main/java/com/google/cloud/teleport/v2/cdc/merge/MergeConfiguration.java
+++ b/v2/common/src/main/java/com/google/cloud/teleport/v2/cdc/merge/MergeConfiguration.java
@@ -19,6 +19,7 @@ import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Prec
 
 import com.google.auto.value.AutoValue;
 import java.io.Serializable;
+import javax.annotation.Nullable;
 import org.joda.time.Duration;
 
 /** Class {@link MergeConfiguration}. */
@@ -64,6 +65,9 @@ public abstract class MergeConfiguration implements Serializable {
   // BigQuery-specific properties
   public static final String BIGQUERY_QUOTE_CHARACTER = "`";
 
+  @Nullable
+  public abstract String projectId();
+
   public abstract String quoteCharacter();
 
   public abstract Boolean supportPartitionedTables();
@@ -82,6 +86,10 @@ public abstract class MergeConfiguration implements Serializable {
     return MergeConfiguration.builder().setQuoteCharacter(BIGQUERY_QUOTE_CHARACTER).build();
   }
 
+  public MergeConfiguration withProjectId(String projectId) {
+    return this.toBuilder().setProjectId(projectId).build();
+  }
+
   public MergeConfiguration withPartitionRetention(int partitionRetention) {
     checkArgument(partitionRetention > 0, "partitionRetention must be greater than 0");
     return this.toBuilder().setPartitionRetention(Integer.valueOf(partitionRetention)).build();
@@ -115,6 +123,8 @@ public abstract class MergeConfiguration implements Serializable {
 
   @AutoValue.Builder
   abstract static class Builder {
+    abstract Builder setProjectId(String projectId);
+
     abstract Builder setQuoteCharacter(String quote);
 
     abstract Builder setSupportPartitionedTables(Boolean supportPartitionedTables);
diff --git a/v2/datastream-to-bigquery/src/main/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQuery.java b/v2/datastream-to-bigquery/src/main/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQuery.java
index b030e8ba..4096c287 100644
--- a/v2/datastream-to-bigquery/src/main/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQuery.java
+++ b/v2/datastream-to-bigquery/src/main/java/com/google/cloud/teleport/v2/templates/DataStreamToBigQuery.java
@@ -571,6 +571,7 @@ public class DataStreamToBigQuery {
               "BigQuery Merge/Merge into Replica Tables",
               BigQueryMerger.of(
                   MergeConfiguration.bigQueryConfiguration()
+                      .withProjectId(bigqueryProjectId)
                       .withMergeWindowDuration(
                           Duration.standardMinutes(options.getMergeFrequencyMinutes()))
                       .withMergeConcurrency(options.getMergeConcurrency())