BugFix for PANW Issue

docs/bigquery-cdcTarget.md

@@ -43,7 +43,7 @@ that the BigQuery job will run in. `BigQuery Job User` role on this project must
 account to run the job. If a temporary bucket needs to be created, the bucket will also be created in this project and
 'GCE Storage Bucket Admin' role on this project must be granted to the specified service account to create buckets.
 
-**Dataset Project ID**: Project the destination dataset belongs to. This is only required if the dataset is not
+**Dataset Project ID**: Project Id for the destination dataset belongs to. This is only required if the dataset is not
 in the same project that the BigQuery job will run in. If no value is given, it will default to the
 configured Project ID. `BigQuery Data Editor` role on this project must be granted to the specified service account to
 write BigQuery data to this project.

src/main/java/io/cdap/delta/bigquery/BigQueryTarget.java

@@ -109,7 +109,7 @@ public void initialize(DeltaTargetContext context) throws Exception {
 
     Credentials credentials = conf.getCredentials();
 
-    String project = conf.getDatasetProject();
+    String project = conf.getProject();
 
     String cmekKey = context.getRuntimeArguments().get(GCP_CMEK_KEY_NAME) != null ?
       context.getRuntimeArguments().get(GCP_CMEK_KEY_NAME) : conf.getEncryptionKeyName();
@@ -138,17 +138,15 @@ public void initialize(DeltaTargetContext context) throws Exception {
             });
     try {
       long maximumExistingSequenceNumber = Failsafe.with(retryPolicy).get(() ->
-              BigQueryUtils.getMaximumExistingSequenceNumber(context.getAllTables(), project, conf.getDatasetName(),
-                      bigQuery, encryptionConfig, MAX_TABLES_PER_QUERY));
+              BigQueryUtils.getMaximumExistingSequenceNumber(context.getAllTables(), conf.getDatasetProject(),
+                conf.getDatasetName(), bigQuery, encryptionConfig, MAX_TABLES_PER_QUERY));
       LOG.info("Found maximum sequence number {}", maximumExistingSequenceNumber);
       context.initializeSequenceNumber(maximumExistingSequenceNumber);
     } catch (Exception e) {
       throw new RuntimeException("Failed to compute the maximum sequence number among all the target tables " +
               "selected for replication. Please make sure that if target tables exists, " +
               "they should have '_sequence_num' column in them.", e);
     }
-
-
   }
 
   @Override

src/main/java/io/cdap/delta/bigquery/BigQueryUtils.java

@@ -19,6 +19,7 @@
 import com.google.cloud.bigquery.BigQuery;
 import com.google.cloud.bigquery.BigQueryError;
 import com.google.cloud.bigquery.BigQueryException;
+import com.google.cloud.bigquery.DatasetId;
 import com.google.cloud.bigquery.EncryptionConfiguration;
 import com.google.cloud.bigquery.FieldValue;
 import com.google.cloud.bigquery.FieldValueList;
@@ -115,8 +116,9 @@ static long getMaximumExistingSequenceNumberPerBatch(Set<SourceTable> allTables,
     SourceTable table0 = allTables.stream().findFirst().get();
     Set<TableId> existingTableIDs = new HashSet<>();
     String dataset = getNormalizedDatasetName(datasetName, table0.getDatabase());
-    if (bigQuery.getDataset(dataset) != null) {
-      for (Table table : bigQuery.listTables(dataset).iterateAll()) {
+    DatasetId datasetId = DatasetId.of(project, dataset);
+    if (bigQuery.getDataset(datasetId) != null) {
+      for (Table table : bigQuery.listTables(datasetId).iterateAll()) {
         existingTableIDs.add(table.getTableId());
       }
     }