Logs the BigQuery error that caused the load job failure (resolves #1… (apache#36068)

* Logs the BigQuery error that caused the load job failure (resolves apache#18419)

* Checks that the BigQuery root cause message to be logged is not empty (resolves apache#18419)

Author: mburuian6

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryHelpers.java

@@ -25,6 +25,7 @@
 import com.google.api.client.util.Sleeper;
 import com.google.api.services.bigquery.model.Clustering;
 import com.google.api.services.bigquery.model.Dataset;
+import com.google.api.services.bigquery.model.ErrorProto;
 import com.google.api.services.bigquery.model.Job;
 import com.google.api.services.bigquery.model.JobReference;
 import com.google.api.services.bigquery.model.JobStatus;
@@ -205,6 +206,7 @@ static class PendingJob implements Serializable {
     void runJob() throws IOException {
       ++currentAttempt;
       if (!shouldRetry()) {
+        logBigQueryError(lastJobAttempted);
         throw new RuntimeException(
             String.format(
                 "Failed to create job with prefix %s, "
@@ -281,6 +283,21 @@ boolean pollJob() throws IOException {
     boolean shouldRetry() {
       return currentAttempt < maxRetries + 1;
     }
+
+    void logBigQueryError(@Nullable Job job) {
+      if (job == null || !parseStatus(job).equals(Status.FAILED)) {
+        return;
+      }
+
+      List<ErrorProto> jobErrors = job.getStatus().getErrors();
+      String finalError = job.getStatus().getErrorResult().getMessage();
+      String causativeError =
+          jobErrors != null && !jobErrors.isEmpty()
+              ? String.format(" due to: %s", jobErrors.get(jobErrors.size() - 1).getMessage())
+              : "";
+
+      LOG.error(String.format("BigQuery Error : %s %s", finalError, causativeError));
+    }
   }
 
   static class RetryJobId implements Serializable {