[BEAM-14405] Fix NPE when ProjectID is not specified in a template ex…

…ecution

PR apache#16547 added handling for null ProjectIDs on command line execution,
but did not handle the possibility of null ProjectIDs in template
execution. PR apache#17094 added nullness checks in MonitoringInfoMetricNames,
which then triggered NPEs if the ProjectID was not specified during a
template execution.

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

@@ -194,7 +194,7 @@ public ReadFromPartitionFn(
     public void setup() throws Exception {
       spannerAccessor = SpannerAccessor.getOrCreate(config);
       projectId =
-          this.config.getProjectId() == null
+          this.config.getProjectId() == null || this.config.getProjectId().get() == null
               ? SpannerOptions.getDefaultProjectId()
               : this.config.getProjectId().get();
     }

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

@@ -1975,6 +1975,7 @@ public void setup() {
 
       projectId =
           this.spannerConfig.getProjectId() == null
+                  || this.spannerConfig.getProjectId().get() == null
               ? SpannerOptions.getDefaultProjectId()
               : this.spannerConfig.getProjectId().get();
     }

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

@@ -208,6 +208,16 @@ public void runReadTestWithDefaultProject() throws Exception {
             .withServiceFactory(serviceFactory));
   }
 
+  @Test
+  public void runReadTestWithNullProject() throws Exception {
+    runReadTest(
+        SpannerConfig.create()
+            .withProjectId((String) null)
+            .withInstanceId("123")
+            .withDatabaseId("aaa")
+            .withServiceFactory(serviceFactory));
+  }
+
   private void runReadTest(SpannerConfig spannerConfig) throws Exception {
     Timestamp timestamp = Timestamp.ofTimeMicroseconds(12345);
     TimestampBound timestampBound = TimestampBound.ofReadTimestamp(timestamp);

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

@@ -323,6 +323,22 @@ public void singleMutationPipelineNoProjectId() throws Exception {
     verifyBatches(batch(m(2L)));
   }
 
+  @Test
+  public void singleMutationPipelineNullProjectId() throws Exception {
+    Mutation mutation = m(2L);
+    PCollection<Mutation> mutations = pipeline.apply(Create.of(mutation));
+
+    mutations.apply(
+        SpannerIO.write()
+            .withProjectId((String) null)
+            .withInstanceId("test-instance")
+            .withDatabaseId("test-database")
+            .withServiceFactory(serviceFactory));
+    pipeline.run();
+
+    verifyBatches(batch(m(2L)));
+  }
+
   @Test
   public void singleMutationGroupPipeline() throws Exception {
     PCollection<MutationGroup> mutations =