fix(spanner): avoid data race on DIRECTPATH_CHANNEL_CREATED by using volatile (#13727)

Fixes b/509632639.

## Problem

`GapicSpannerRpc` declared a process-wide mutable static flag:

```java
public static boolean DIRECTPATH_CHANNEL_CREATED = false;
```

The flag is written from the `GapicSpannerRpc` constructor without
synchronization and read by `HeaderInterceptor` when populating the
`directpath_enabled` built-in metrics attribute. When multiple Spanner
clients are constructed concurrently (for example Apache Beam's
DirectRunner initializing `DoFn`s on multiple worker threads), the
unsynchronized write/read pair is a data race; ThreadSanitizer halts the
JVM on it.

## Fix

Declare the field `volatile`:

```java
public static volatile boolean DIRECTPATH_CHANNEL_CREATED = false;
```

- The volatile write in the constructor happens-before every subsequent
volatile read in `HeaderInterceptor`, eliminating the data race. The
flag has no read-modify-write usage, so `volatile` is sufficient; the
existing last-writer-wins semantics are preserved unchanged.
- `volatile` was chosen over `AtomicBoolean` deliberately: it keeps the
field's source and binary signature (`boolean`) intact, so any external
code reading `GapicSpannerRpc.DIRECTPATH_CHANNEL_CREATED` keeps
compiling and linking. (The class is `@InternalApi` and the `spi.v1`
package is excluded from clirr checks, but there is no reason to break
the field's ABI when `volatile` fixes the race equally well.)
- Javadoc on the field documents the concurrency semantics.

Author: rahul2393

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java

@@ -255,7 +255,15 @@ public class GapicSpannerRpc implements SpannerRpc {
   private static final String CLIENT_LIBRARY_LANGUAGE = "spanner-java";
   public static final String DEFAULT_USER_AGENT =
       CLIENT_LIBRARY_LANGUAGE + "/" + GaxProperties.getLibraryVersion(GapicSpannerRpc.class);
-  public static boolean DIRECTPATH_CHANNEL_CREATED = false;
+
+  /**
+   * Whether the most recently initialized RPC created a DirectPath channel.
+   *
+   * <p>This process-wide volatile value may be updated during concurrent client construction and
+   * read from another thread when built-in metric attributes are created.
+   */
+  public static volatile boolean DIRECTPATH_CHANNEL_CREATED = false;
+
   private static final String API_FILE = "grpc-gcp-apiconfig.json";
 
   private final RequestIdCreator requestIdCreator = new RequestIdCreatorImpl();

java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpcTest.java

@@ -106,14 +106,20 @@
 import java.lang.reflect.Modifier;
 import java.net.InetSocketAddress;
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.IdentityHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -934,6 +940,49 @@ public void testAdminStubSettings_whenStubNotInitialized_assertNullClientSetting
     rpc.shutdown();
   }
 
+  @Test
+  public void testConcurrentClientCreationDoesNotRaceOnDirectPathFlag() throws Exception {
+    // Concurrent creation of Spanner clients used to cause a data race on the static
+    // DIRECTPATH_CHANNEL_CREATED field, which was written from the constructor without
+    // synchronization. This verifies that concurrent client creation succeeds and leaves the flag
+    // in a consistent state now that the field is volatile.
+    int numThreads = 8;
+    ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+    CountDownLatch start = new CountDownLatch(1);
+    List<Future<Void>> futures = new ArrayList<>(numThreads);
+    try {
+      for (int i = 0; i < numThreads; i++) {
+        futures.add(
+            executor.submit(
+                () -> {
+                  start.await();
+                  GapicSpannerRpc rpc = new GapicSpannerRpc(createSpannerOptions(), true);
+                  try {
+                    return null;
+                  } finally {
+                    rpc.shutdown();
+                  }
+                }));
+      }
+      start.countDown();
+      for (Future<Void> future : futures) {
+        future.get(60L, TimeUnit.SECONDS);
+      }
+      // The test options connect to a local plaintext mock server, so no DirectPath channel is
+      // ever created.
+      assertTrue(
+          Modifier.isVolatile(
+              GapicSpannerRpc.class.getField("DIRECTPATH_CHANNEL_CREATED").getModifiers()));
+      assertFalse(GapicSpannerRpc.DIRECTPATH_CHANNEL_CREATED);
+    } finally {
+      start.countDown();
+      for (Future<Void> future : futures) {
+        future.cancel(true);
+      }
+      executor.shutdownNow();
+    }
+  }
+
   @Test
   public void testCreateSession_assertSessionProto() {
     SpannerOptions options = createSpannerOptions();