fix(bigtable): address PR review comments on ReadModifyWriteRow shim

- Remove stale TODO comments from TableAsync.readModifyWriteRow
- Rephrase "must never be retried" to "should not be retried unless
  directed by server" for CheckAndMutateRow and ReadModifyWriteRow in
  TableBase
- Extract shared buildRow logic into RowConverter utility, used by both
  ReadRowShim and ReadModifyWriteRowShim
- Remove CheckAndMutateRowShim and ReadModifyWriteRowShimInner from
  ShimImpl until session routing is enabled; decorator methods now
  return the classic callable directly to avoid creating unused session
  pools

Change-Id: I7a803404cbf5c945b7e03aac4c970f0ed31feec8

Author: mutianf

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableAsync.java

@@ -133,7 +133,6 @@ public CompletableFuture<SessionCheckAndMutateRowResponse> checkAndMutateRow(
     return f;
   }
 
-  // TODO: get deadline from compatibility layer
   public CompletableFuture<SessionReadModifyWriteRowResponse> readModifyWriteRow(
       SessionReadModifyWriteRowRequest req, Deadline deadline) {
     UnaryResponseFuture<SessionReadModifyWriteRowResponse> f = new UnaryResponseFuture<>();

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableBase.java

@@ -181,7 +181,7 @@ public void checkAndMutateRow(
         new RetryingVRpc<>(() -> sessionPool.newCall(checkAndMutateRowDescriptor), timer);
     VRpcTracer tracer =
         metrics.newTableTracer(sessionPool.getInfo(), checkAndMutateRowDescriptor, deadline);
-    // CheckAndMutateRow is not idempotent and must never be retried.
+    // CheckAndMutateRow is not idempotent and should not be retried unless directed by server.
     new VOperationImpl<>(retry, Context.current(), userCallbackExecutor, tracer, deadline, false)
         .start(req, listener);
   }
@@ -194,7 +194,7 @@ public void readModifyWriteRow(
         new RetryingVRpc<>(() -> sessionPool.newCall(readModifyWriteRowDescriptor), timer);
     VRpcTracer tracer =
         metrics.newTableTracer(sessionPool.getInfo(), readModifyWriteRowDescriptor, deadline);
-    // ReadModifyWriteRow is not idempotent and must never be retried.
+    // ReadModifyWriteRow is not idempotent and should not be retried unless directed by server.
     new VOperationImpl<>(retry, Context.current(), userCallbackExecutor, tracer, deadline, false)
         .start(req, listener);
   }

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

@@ -34,11 +34,8 @@
 import com.google.cloud.bigtable.data.v2.internal.api.Client;
 import com.google.cloud.bigtable.data.v2.internal.api.Client.Resource;
 import com.google.cloud.bigtable.data.v2.internal.channels.ChannelPool;
-import com.google.cloud.bigtable.data.v2.internal.compat.ops.CheckAndMutateRowShim;
 import com.google.cloud.bigtable.data.v2.internal.compat.ops.DivertingUnaryCallable;
 import com.google.cloud.bigtable.data.v2.internal.compat.ops.MutateRowShim;
-import com.google.cloud.bigtable.data.v2.internal.compat.ops.ReadModifyWriteRowShim;
-import com.google.cloud.bigtable.data.v2.internal.compat.ops.ReadModifyWriteRowShimInner;
 import com.google.cloud.bigtable.data.v2.internal.compat.ops.ReadRowShim;
 import com.google.cloud.bigtable.data.v2.internal.compat.ops.ReadRowShimInner;
 import com.google.cloud.bigtable.data.v2.internal.csm.Metrics;
@@ -91,8 +88,6 @@ public class ShimImpl implements Shim {
 
   private final ReadRowShimInner readRowShimInner;
   private final MutateRowShim mutateRowShim;
-  private final CheckAndMutateRowShim checkAndMutateRowShim;
-  private final ReadModifyWriteRowShimInner readModifyWriteRowShimInner;
 
   public static Shim create(
       ClientInfo clientInfo,
@@ -208,8 +203,6 @@ public ShimImpl(
 
     this.readRowShimInner = new ReadRowShimInner(client);
     this.mutateRowShim = new MutateRowShim(client);
-    this.checkAndMutateRowShim = new CheckAndMutateRowShim(client);
-    this.readModifyWriteRowShimInner = new ReadModifyWriteRowShimInner(client);
   }
 
   /**
@@ -388,24 +381,16 @@ public UnaryCallable<RowMutation, Void> decorateMutateRow(
   @Override
   public UnaryCallable<ConditionalRowMutation, Boolean> decorateCheckAndMutateRow(
       UnaryCallable<ConditionalRowMutation, Boolean> classic, UnaryCallSettings<?, ?> settings) {
-    return new DivertingUnaryCallable<>(
-        configManager,
-        classic,
-        checkAndMutateRowShim,
-        Util.extractTimeout(settings),
-        debugTagTracer);
+    // TODO: enable session routing once diversion by method is added
+    return classic;
   }
 
   @Override
   public <RowT> UnaryCallable<ReadModifyWriteRow, RowT> decorateReadModifyWriteRow(
       UnaryCallable<ReadModifyWriteRow, RowT> classic,
       RowAdapter<RowT> rowAdapter,
       UnaryCallSettings<?, ?> settings) {
-    return new DivertingUnaryCallable<>(
-        configManager,
-        classic,
-        new ReadModifyWriteRowShim<>(readModifyWriteRowShimInner, rowAdapter),
-        Util.extractTimeout(settings),
-        debugTagTracer);
+    // TODO: enable session routing once server-side changes are rolled out
+    return classic;
   }
 }

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

@@ -15,9 +15,6 @@
  */
 package com.google.cloud.bigtable.data.v2.internal.compat.ops;
 
-import com.google.bigtable.v2.Cell;
-import com.google.bigtable.v2.Column;
-import com.google.bigtable.v2.Family;
 import com.google.bigtable.v2.SessionReadModifyWriteRowResponse;
 import com.google.cloud.bigtable.data.v2.models.DefaultRowAdapter;
 import com.google.cloud.bigtable.data.v2.models.ReadModifyWriteRow;
@@ -68,24 +65,6 @@ private static <T> T buildRow(RowBuilder<T> adapter, SessionReadModifyWriteRowRe
     if (!input.hasRow()) {
       return null;
     }
-    com.google.bigtable.v2.Row protoRow = input.getRow();
-
-    adapter.startRow(protoRow.getKey());
-
-    for (Family family : protoRow.getFamiliesList()) {
-      for (Column column : family.getColumnsList()) {
-        for (Cell cell : column.getCellsList()) {
-          adapter.startCell(
-              family.getName(),
-              column.getQualifier(),
-              cell.getTimestampMicros(),
-              cell.getLabelsList(),
-              0);
-          adapter.cellValue(cell.getValue());
-          adapter.finishCell();
-        }
-      }
-    }
-    return adapter.finishRow();
+    return RowConverter.buildRow(adapter, input.getRow());
   }
 }

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

@@ -15,9 +15,6 @@
  */
 package com.google.cloud.bigtable.data.v2.internal.compat.ops;
 
-import com.google.bigtable.v2.Cell;
-import com.google.bigtable.v2.Column;
-import com.google.bigtable.v2.Family;
 import com.google.bigtable.v2.SessionReadRowResponse;
 import com.google.cloud.bigtable.data.v2.models.DefaultRowAdapter;
 import com.google.cloud.bigtable.data.v2.models.Query;
@@ -67,24 +64,6 @@ private static <T> T buildRow(RowBuilder<T> adapter, SessionReadRowResponse inpu
     if (!input.hasRow()) {
       return null;
     }
-    com.google.bigtable.v2.Row protoRow = input.getRow();
-
-    adapter.startRow(protoRow.getKey());
-
-    for (Family family : protoRow.getFamiliesList()) {
-      for (Column column : family.getColumnsList()) {
-        for (Cell cell : column.getCellsList()) {
-          adapter.startCell(
-              family.getName(),
-              column.getQualifier(),
-              cell.getTimestampMicros(),
-              cell.getLabelsList(),
-              0);
-          adapter.cellValue(cell.getValue());
-          adapter.finishCell();
-        }
-      }
-    }
-    return adapter.finishRow();
+    return RowConverter.buildRow(adapter, input.getRow());
   }
 }

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

@@ -0,0 +1,47 @@
+/*
+ * 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
+ *
+ *     https://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.bigtable.data.v2.internal.compat.ops;
+
+import com.google.bigtable.v2.Cell;
+import com.google.bigtable.v2.Column;
+import com.google.bigtable.v2.Family;
+import com.google.bigtable.v2.Row;
+import com.google.cloud.bigtable.data.v2.models.RowAdapter.RowBuilder;
+
+class RowConverter {
+
+  private RowConverter() {}
+
+  static <T> T buildRow(RowBuilder<T> builder, Row protoRow) {
+    builder.startRow(protoRow.getKey());
+
+    for (Family family : protoRow.getFamiliesList()) {
+      for (Column column : family.getColumnsList()) {
+        for (Cell cell : column.getCellsList()) {
+          builder.startCell(
+              family.getName(),
+              column.getQualifier(),
+              cell.getTimestampMicros(),
+              cell.getLabelsList(),
+              0);
+          builder.cellValue(cell.getValue());
+          builder.finishCell();
+        }
+      }
+    }
+    return builder.finishRow();
+  }
+}