fix: route SessionPoolMap.apply op throws to failed futures

`apply` already converted lookup throws to failed futures, but a sync throw from
`op.apply(v)` itself (NPE on malformed input, REE during shutdown, any RuntimeException
from the wrapped call chain) propagated to the caller — contradicting the docstring's
contract that "callers consistently observe failures through the future surface."

Wrap the op invocation in try/catch with the same pattern as the lookup arm.

Author: igorbernstein2

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/compat/ops/SessionPoolMap.java

@@ -54,9 +54,10 @@ public V get(K key) {
   }
 
   /**
-   * Looks up the cached value and applies {@code op}. If the lookup throws (e.g. {@link
-   * IllegalStateException} from a closed Client), the throw is converted to a failed future so
-   * callers consistently observe failures through the future surface.
+   * Looks up the cached value and applies {@code op}. Both the lookup (e.g. {@link
+   * IllegalStateException} from a closed Client) and a synchronous throw from {@code op.apply}
+   * (e.g. NPE on malformed input, RejectedExecutionException during shutdown) are converted to a
+   * failed future so callers consistently observe failures through the future surface.
    */
   public <R> CompletableFuture<R> apply(K key, Function<V, CompletableFuture<R>> op) {
     V v;
@@ -67,7 +68,13 @@ public <R> CompletableFuture<R> apply(K key, Function<V, CompletableFuture<R>> o
       f.completeExceptionally(e);
       return f;
     }
-    return op.apply(v);
+    try {
+      return op.apply(v);
+    } catch (Throwable t) {
+      CompletableFuture<R> f = new CompletableFuture<>();
+      f.completeExceptionally(t);
+      return f;
+    }
   }
 
   /** Evicts all entries, triggering Closeable.close on each via the cache's removal listener. */

java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/compat/ops/SessionPoolMapTest.java

@@ -72,6 +72,28 @@ void apply_convertsLoaderThrowToFailedFuture() throws Exception {
     assertThat(ee).hasCauseThat().hasMessageThat().isEqualTo("Client is closed");
   }
 
+  @Test
+  void apply_convertsOpSyncThrowToFailedFuture() throws Exception {
+    // op.apply may throw synchronously (NPE on malformed input, REE during shutdown, any
+    // RuntimeException from the wrapped call chain). The async surface must not propagate
+    // the throw — same contract as the loader-throw path above.
+    CountingHandle handle = new CountingHandle();
+    SessionPoolMap<String, CountingHandle> map = new SessionPoolMap<>(key -> handle);
+
+    CompletableFuture<String> result =
+        map.apply(
+            "k",
+            v -> {
+              throw new IllegalStateException("op blew up");
+            });
+
+    assertThat(result.isCompletedExceptionally()).isTrue();
+    ExecutionException ee =
+        assertThrows(ExecutionException.class, () -> result.get(1, TimeUnit.SECONDS));
+    assertThat(ee).hasCauseThat().isInstanceOf(IllegalStateException.class);
+    assertThat(ee).hasCauseThat().hasMessageThat().isEqualTo("op blew up");
+  }
+
   @Test
   void apply_happyPathInvokesOp() throws Exception {
     CountingHandle handle = new CountingHandle();