use ThreadLocal so that metrics can re-use this solution if span is turned off

# Conflicts:
#	java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryRetryHelper.java

Author: ldetmer

java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryRetryHelper.java

@@ -27,21 +27,15 @@
 import com.google.cloud.RetryHelper;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.Tracer;
-import io.opentelemetry.context.Context;
-import io.opentelemetry.context.ContextKey;
 import io.opentelemetry.context.Scope;
 import java.io.IOException;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
 public class BigQueryRetryHelper extends RetryHelper {
 
-  public static final ContextKey<AtomicInteger> RETRY_ATTEMPT_KEY =
-      ContextKey.named("bq_retry_attempt");
-
   private static final Logger LOG = Logger.getLogger(BigQueryRetryHelper.class.getName());
 
   public static <V> V runWithRetries(
@@ -60,11 +54,8 @@ public static <V> V runWithRetries(
               .spanBuilder("com.google.cloud.bigquery.BigQueryRetryHelper.runWithRetries")
               .startSpan();
     }
-    Context retryContext = Context.current().with(RETRY_ATTEMPT_KEY, new AtomicInteger(0));
-    if (runWithRetries != null) {
-      retryContext = retryContext.with(runWithRetries);
-    }
-    try (Scope runWithRetriesScope = retryContext.makeCurrent()) {
+    try (Scope runWithRetriesScope = runWithRetries != null ? runWithRetries.makeCurrent() : null;
+        BigQueryRetryTracker tracker = new BigQueryRetryTracker()) {
       // Suppressing should be ok as a workaraund. Current and only ResultRetryAlgorithm
       // implementation does not use response at all, so ignoring its type is ok.
       @SuppressWarnings("unchecked")

java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryRetryTracker.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigquery;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Tracks retry attempts for a client request using a ThreadLocal for telemetry data.
+ *
+ * <p>It implements {@link AutoCloseable} to ensure that the ThreadLocal is safely restored to its
+ * previous state (or removed) when the try-with-resources block exits, preventing memory leaks and
+ * state bleeding in thread-pooling environments.
+ */
+public class BigQueryRetryTracker implements AutoCloseable {
+
+  private static final ThreadLocal<AtomicInteger> HOLDER = new ThreadLocal<>();
+  private final AtomicInteger previous;
+
+  public BigQueryRetryTracker() {
+    this.previous = HOLDER.get();
+    HOLDER.set(new AtomicInteger(0));
+  }
+
+  public static AtomicInteger get() {
+    return HOLDER.get();
+  }
+
+  @Override
+  public void close() {
+    if (previous == null) {
+      HOLDER.remove();
+    } else {
+      HOLDER.set(previous);
+    }
+  }
+}

java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/telemetry/HttpTracingRequestInitializer.java

@@ -19,7 +19,7 @@
 import com.google.api.client.http.*;
 import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
-import com.google.cloud.bigquery.BigQueryRetryHelper;
+import com.google.cloud.bigquery.BigQueryRetryTracker;
 import com.google.common.annotations.VisibleForTesting;
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.trace.Span;
@@ -86,7 +86,7 @@ public void initialize(HttpRequest request) throws IOException {
 
     addInitialHttpAttributesToSpan(span, request);
 
-    AtomicInteger attemptTracker = Context.current().get(BigQueryRetryHelper.RETRY_ATTEMPT_KEY);
+    AtomicInteger attemptTracker = BigQueryRetryTracker.get();
     if (attemptTracker != null) {
       int attempt = attemptTracker.getAndIncrement();
       if (attempt > 0) {

java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/BigQueryRetryTrackerTest.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigquery;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Test;
+
+public class BigQueryRetryTrackerTest {
+
+  @Test
+  public void testCreateAndClose() {
+    assertNull(BigQueryRetryTracker.get(), "Tracker should be null before create");
+
+    try (BigQueryRetryTracker tracker = new BigQueryRetryTracker()) {
+      AtomicInteger counter = BigQueryRetryTracker.get();
+      assertNotNull(counter);
+      assertEquals(0, counter.get());
+    }
+
+    assertNull(BigQueryRetryTracker.get(), "Tracker should be cleaned up after close");
+  }
+
+  @Test
+  public void testThreadIsolation() throws ExecutionException, InterruptedException {
+    CompletableFuture<Void> future1 =
+        CompletableFuture.runAsync(
+            () -> {
+              try (BigQueryRetryTracker tracker = new BigQueryRetryTracker()) {
+                BigQueryRetryTracker.get().set(10);
+                try {
+                  Thread.sleep(100);
+                } catch (InterruptedException e) {
+                  Thread.currentThread().interrupt();
+                }
+                assertEquals(10, BigQueryRetryTracker.get().get());
+              }
+            });
+
+    CompletableFuture<Void> future2 =
+        CompletableFuture.runAsync(
+            () -> {
+              try (BigQueryRetryTracker tracker = new BigQueryRetryTracker()) {
+                BigQueryRetryTracker.get().set(20);
+                assertEquals(20, BigQueryRetryTracker.get().get());
+              }
+            });
+
+    future1.get();
+    future2.get();
+  }
+}

java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/telemetry/HttpTracingRequestInitializerTest.java

@@ -44,11 +44,10 @@
 import com.google.api.client.testing.http.MockHttpTransport;
 import com.google.api.client.testing.http.MockLowLevelHttpRequest;
 import com.google.api.client.testing.http.MockLowLevelHttpResponse;
-import com.google.cloud.bigquery.BigQueryRetryHelper;
+import com.google.cloud.bigquery.BigQueryRetryTracker;
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.Tracer;
-import io.opentelemetry.context.Context;
 import io.opentelemetry.context.Scope;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
 import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
@@ -57,7 +56,6 @@
 import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
 import java.io.IOException;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -321,15 +319,13 @@ public void testAddRequestBodySizeToSpan_WithEncoding() throws IOException {
   @Test
   public void testRetryCountFromContext() throws IOException {
     HttpTransport transport = createTransport();
-    AtomicInteger counter = new AtomicInteger(2);
-    Context context =
-        io.opentelemetry.context.Context.current()
-            .with(BigQueryRetryHelper.RETRY_ATTEMPT_KEY, counter);
 
-    try (io.opentelemetry.context.Scope scope = context.makeCurrent()) {
+    try (BigQueryRetryTracker tracker = new BigQueryRetryTracker()) {
+      BigQueryRetryTracker.get().set(2);
       HttpRequest request = buildGetRequest(transport, initializer, BASE_URL);
       HttpResponse response = request.execute();
       response.disconnect();
+      assertEquals(3, BigQueryRetryTracker.get().get());
     }
 
     spanScope.close();
@@ -340,7 +336,6 @@ public void testRetryCountFromContext() throws IOException {
     SpanData span = spans.get(0);
     assertEquals(
         2L, span.getAttributes().get(HttpTracingRequestInitializer.HTTP_REQUEST_RESEND_COUNT));
-    assertEquals(3, counter.get());
   }
 
   @Test