Fix Null Pointer Exception in BigQueryIO (#36209)

Author: tarun-google

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

@@ -285,7 +285,7 @@ public void validate(@Nullable PipelineOptions maybeOptions) {
     PipelineOptions options = Preconditions.checkArgumentNotNull(maybeOptions);
     // We will use a BigQuery load job -- validate the temp location.
     String tempLocation;
-    if (customGcsTempLocation == null) {
+    if (customGcsTempLocation == null || customGcsTempLocation.get() == null) {
       tempLocation = options.getTempLocation();
     } else {
       if (!customGcsTempLocation.isAccessible()) {
@@ -589,7 +589,7 @@ private PCollectionView<String> createTempFilePrefixView(
                       @ProcessElement
                       public void getTempFilePrefix(ProcessContext c) {
                         String tempLocationRoot;
-                        if (customGcsTempLocation != null) {
+                        if (customGcsTempLocation != null && customGcsTempLocation.get() != null) {
                           tempLocationRoot = customGcsTempLocation.get();
                         } else {
                           tempLocationRoot = c.getPipelineOptions().getTempLocation();

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

@@ -4441,4 +4441,29 @@ public void testUpsertAndDeleteBeamRows() throws Exception {
         fakeDatasetService.getAllRows("project-id", "dataset-id", "table-id"),
         containsInAnyOrder(Iterables.toArray(expected, TableRow.class)));
   }
+
+  @Test
+  public void testCustomGcsTempLocationNull() throws Exception {
+    BigQueryIO.Write<TableRow> write =
+        BigQueryIO.writeTableRows()
+            .to("dataset-id.table-id")
+            .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
+            .withSchema(
+                new TableSchema()
+                    .setFields(
+                        ImmutableList.of(new TableFieldSchema().setName("name").setType("STRING"))))
+            .withMethod(Method.FILE_LOADS)
+            .withoutValidation()
+            .withTestServices(fakeBqServices)
+            .withCustomGcsTempLocation(ValueProvider.StaticValueProvider.of(null));
+
+    p.apply(
+            Create.of(new TableRow().set("name", "a"), new TableRow().set("name", "b"))
+                .withCoder(TableRowJsonCoder.of()))
+        .apply("WriteToBQ", write);
+    p.run();
+    assertThat(
+        fakeDatasetService.getAllRows("project-id", "dataset-id", "table-id"),
+        containsInAnyOrder(new TableRow().set("name", "a"), new TableRow().set("name", "b")));
+  }
 }