fix: improve transaction metrics correctness and coverage

- Fix ReadWriteTransactionCallable.call() to use transaction.isActive()
  as the sole rollback guard in both the catch and finally blocks,
  removing the redundant `committed` boolean flag. isActive() returns
  false after both a successful commit and a completed rollback, making
  it sufficient to prevent double-rollback or rollback-after-commit.
  Move recordAttempt() to the finally block so it always runs exactly
  once regardless of how the attempt exits.

- Remove transaction metrics from AggregationQueryExecutor. Per-query
  metrics were recorded outside the runInTransaction() scope and used
  the misleading transaction_latency name for a single RPC latency.
  Metrics are now scoped exclusively to runInTransaction() flows.

- Differentiate method attributes: transaction_latency uses
  Transaction.Run (overall operation) and transaction_attempt_count
  uses Transaction.Commit (per-attempt), reflecting their different
  scopes.

- Add project_id and database_id attributes to both metrics for
  multi-project and multi-database observability.

- Move Stopwatch start to just before RetryHelper.runWithRetries() to
  exclude callable construction overhead from latency measurements.

- Fix transaction_latency description from "successful transaction
  operations" to "transaction operations" since the metric is recorded
  for both successful and failed transactions.

- Add Javadoc note to extractStatusCode() about the null-reason edge
  case where UNKNOWN is returned.

- Add latency-on-failure assertion to DatastoreImplMetricsTest and
  extend attribute checks to cover project_id and database_id.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: lqiu96

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/DatastoreException.java

@@ -168,6 +168,11 @@ static DatastoreException propagateUserException(Exception ex) {
    * com.google.api.gax.rpc.StatusCode.Code} which is transport-neutral, supporting both gRPC and
    * HttpJson. Falls back to "UNKNOWN" if the status cannot be determined.
    *
+   * <p>Note: Some {@link DatastoreException} instances are constructed without a reason (e.g. via
+   * {@link DatastoreException#DatastoreException(int, String, Throwable)}). If all {@link
+   * DatastoreException} instances in the cause chain have a null or empty reason, this method
+   * returns "UNKNOWN" even if the underlying error carries a meaningful status.
+   *
    * @param throwable the throwable to extract the status code from
    * @return the status code name, or "UNKNOWN" if not determinable
    */

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/DatastoreImpl.java

@@ -112,8 +112,7 @@ final class DatastoreImpl extends BaseService<DatastoreOptions> implements Datas
         new AggregationQueryExecutor(
             new RetryAndTraceDatastoreRpcDecorator(
                 datastoreRpc, otelTraceUtil, retrySettings, options),
-            options,
-            metricsRecorder);
+            options);
   }
 
   @Override
@@ -213,17 +212,28 @@ void setPrevTransactionId(ByteString transactionId) {
 
     @Override
     public T call() throws DatastoreException {
+      String attemptStatus = StatusCode.Code.UNKNOWN.toString();
       try {
         transaction = datastore.newTransaction(options);
         T value = callable.run(transaction);
         transaction.commit();
-        recordAttempt(Status.Code.OK.toString());
+        attemptStatus = Status.Code.OK.toString();
         return value;
       } catch (Exception ex) {
-        transaction.rollback();
-        recordAttempt(DatastoreException.extractStatusCode(ex));
+        attemptStatus = DatastoreException.extractStatusCode(ex);
+        // An exception here can originate from either callable.run() (before commit was attempted)
+        // or from transaction.commit() itself. In both cases the transaction is still active.
+        // isActive() returns false if the transaction was already committed or rolled back, so
+        // it is safe to use as the sole guard here without tracking a separate committed flag.
+        if (transaction.isActive()) {
+          transaction.rollback();
+        }
         throw DatastoreException.propagateUserException(ex);
       } finally {
+        recordAttempt(attemptStatus);
+        // transaction.isActive() returns false after both a successful commit or a completed
+        // rollback, so it already guards against rolling back a committed transaction or
+        // rolling back a transaction that has already been rolled back.
         if (transaction.isActive()) {
           transaction.rollback();
         }
@@ -243,6 +253,10 @@ private void recordAttempt(String status) {
       attributes.put(TelemetryConstants.ATTRIBUTES_KEY_STATUS, status);
       attributes.put(
           TelemetryConstants.ATTRIBUTES_KEY_METHOD, TelemetryConstants.METHOD_TRANSACTION_COMMIT);
+      attributes.put(
+          TelemetryConstants.ATTRIBUTES_KEY_PROJECT_ID, datastore.getOptions().getProjectId());
+      attributes.put(
+          TelemetryConstants.ATTRIBUTES_KEY_DATABASE_ID, datastore.getOptions().getDatabaseId());
       metricsRecorder.recordTransactionAttemptCount(1, attributes);
     }
   }
@@ -256,7 +270,6 @@ public <T> T runInTransaction(final TransactionCallable<T> callable) {
   public <T> T runInTransaction(
       final TransactionCallable<T> callable, TransactionOptions transactionOptions) {
     TraceUtil.Span span = otelTraceUtil.startSpan(SPAN_NAME_TRANSACTION_RUN);
-    Stopwatch stopwatch = Stopwatch.createStarted();
 
     ReadWriteTransactionCallable<T> baseCallable =
         new ReadWriteTransactionCallable<>(this, callable, transactionOptions, metricsRecorder);
@@ -267,6 +280,7 @@ public <T> T runInTransaction(
     }
 
     String status = StatusCode.Code.OK.toString();
+    Stopwatch stopwatch = Stopwatch.createStarted();
     try (Scope ignored = span.makeCurrent()) {
       return RetryHelper.runWithRetries(
           transactionCallable,
@@ -282,7 +296,9 @@ public <T> T runInTransaction(
       Map<String, String> attributes = new HashMap<>();
       attributes.put(TelemetryConstants.ATTRIBUTES_KEY_STATUS, status);
       attributes.put(
-          TelemetryConstants.ATTRIBUTES_KEY_METHOD, TelemetryConstants.METHOD_TRANSACTION_COMMIT);
+          TelemetryConstants.ATTRIBUTES_KEY_METHOD, TelemetryConstants.METHOD_TRANSACTION_RUN);
+      attributes.put(TelemetryConstants.ATTRIBUTES_KEY_PROJECT_ID, getOptions().getProjectId());
+      attributes.put(TelemetryConstants.ATTRIBUTES_KEY_DATABASE_ID, getOptions().getDatabaseId());
       metricsRecorder.recordTransactionLatency(latencyMs, attributes);
       span.end();
     }

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/execution/AggregationQueryExecutor.java

@@ -16,26 +16,18 @@
 package com.google.cloud.datastore.execution;
 
 import com.google.api.core.InternalApi;
-import com.google.api.gax.rpc.StatusCode;
 import com.google.cloud.datastore.AggregationQuery;
 import com.google.cloud.datastore.AggregationResults;
-import com.google.cloud.datastore.DatastoreException;
 import com.google.cloud.datastore.DatastoreOptions;
 import com.google.cloud.datastore.ReadOption;
 import com.google.cloud.datastore.ReadOption.QueryConfig;
 import com.google.cloud.datastore.execution.request.AggregationQueryRequestProtoPreparer;
 import com.google.cloud.datastore.execution.response.AggregationQueryResponseTransformer;
 import com.google.cloud.datastore.models.ExplainOptions;
 import com.google.cloud.datastore.spi.v1.DatastoreRpc;
-import com.google.cloud.datastore.telemetry.MetricsRecorder;
-import com.google.cloud.datastore.telemetry.TelemetryConstants;
-import com.google.common.base.Stopwatch;
 import com.google.datastore.v1.RunAggregationQueryRequest;
 import com.google.datastore.v1.RunAggregationQueryResponse;
 import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
 
 /**
  * An implementation of {@link QueryExecutor} which executes {@link AggregationQuery} and returns
@@ -48,16 +40,13 @@ public class AggregationQueryExecutor
   private final DatastoreRpc datastoreRpc;
   private final AggregationQueryRequestProtoPreparer protoPreparer;
   private final AggregationQueryResponseTransformer responseTransformer;
-  private final MetricsRecorder metricsRecorder;
 
   public AggregationQueryExecutor(
       DatastoreRpc datastoreRpc,
-      DatastoreOptions datastoreOptions,
-      MetricsRecorder metricsRecorder) {
+      DatastoreOptions datastoreOptions) {
     this.datastoreRpc = datastoreRpc;
     this.protoPreparer = new AggregationQueryRequestProtoPreparer(datastoreOptions);
     this.responseTransformer = new AggregationQueryResponseTransformer();
-    this.metricsRecorder = metricsRecorder;
   }
 
   @Override
@@ -66,28 +55,9 @@ public AggregationResults execute(
     RunAggregationQueryRequest runAggregationQueryRequest =
         getRunAggregationQueryRequest(
             query, explainOptions == null ? null : explainOptions.toPb(), readOptions);
-    boolean isTransactional = runAggregationQueryRequest.getReadOptions().hasTransaction();
-    Stopwatch stopwatch = isTransactional ? Stopwatch.createStarted() : null;
-    String status = StatusCode.Code.OK.toString();
-    try {
-      RunAggregationQueryResponse runAggregationQueryResponse =
-          this.datastoreRpc.runAggregationQuery(runAggregationQueryRequest);
-      return this.responseTransformer.transform(runAggregationQueryResponse);
-    } catch (DatastoreException e) {
-      status = DatastoreException.extractStatusCode(e);
-      throw e;
-    } finally {
-      if (isTransactional) {
-        long latencyMs = stopwatch.elapsed(TimeUnit.MILLISECONDS);
-        Map<String, String> attributes = new HashMap<>();
-        attributes.put(TelemetryConstants.ATTRIBUTES_KEY_STATUS, status);
-        attributes.put(
-            TelemetryConstants.ATTRIBUTES_KEY_METHOD,
-            TelemetryConstants.METHOD_TRANSACTION_RUN_AGGREGATION_QUERY);
-        metricsRecorder.recordTransactionLatency(latencyMs, attributes);
-        metricsRecorder.recordTransactionAttemptCount(1, attributes);
-      }
-    }
+    RunAggregationQueryResponse runAggregationQueryResponse =
+        this.datastoreRpc.runAggregationQuery(runAggregationQueryRequest);
+    return this.responseTransformer.transform(runAggregationQueryResponse);
   }
 
   private RunAggregationQueryRequest getRunAggregationQueryRequest(

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/telemetry/OpenTelemetryMetricsRecorder.java

@@ -42,7 +42,7 @@ class OpenTelemetryMetricsRecorder implements MetricsRecorder {
     this.transactionLatency =
         meter
             .histogramBuilder(TelemetryConstants.SERVICE_NAME + "/transaction_latency")
-            .setDescription("Total latency for successful transaction operations")
+            .setDescription("Total latency of transaction operations")
             .setUnit("ms")
             .build();
 

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/telemetry/TelemetryConstants.java

@@ -39,6 +39,12 @@ public class TelemetryConstants {
   /** Attribute key for the RPC method name (e.g. "Transaction.Run"). */
   public static final String ATTRIBUTES_KEY_METHOD = "method";
 
+  /** Attribute key for the GCP project ID. */
+  public static final String ATTRIBUTES_KEY_PROJECT_ID = "project_id";
+
+  /** Attribute key for the Datastore database ID. */
+  public static final String ATTRIBUTES_KEY_DATABASE_ID = "database_id";
+
   /* TODO(lawrenceqiu): For now, these are a duplicate of method names in TraceUtil. Those will use these eventually */
   // Format is not SnakeCase to keep backward compatibility with the existing values TraceUtil spans
   public static final String METHOD_ALLOCATE_IDS = "AllocateIds";

java-datastore/google-cloud-datastore/src/test/java/com/google/cloud/datastore/DatastoreImplMetricsTest.java

@@ -113,7 +113,10 @@ public void runInTransaction_recordsLatencyOnSuccess() {
     assertThat(point.getSum()).isAtLeast(0.0);
     assertThat(point.getAttributes().get(AttributeKey.stringKey("status"))).isEqualTo("OK");
     assertThat(point.getAttributes().get(AttributeKey.stringKey("method")))
-        .isEqualTo(TelemetryConstants.METHOD_TRANSACTION_COMMIT);
+        .isEqualTo(TelemetryConstants.METHOD_TRANSACTION_RUN);
+    assertThat(point.getAttributes().get(AttributeKey.stringKey("project_id")))
+        .isEqualTo(PROJECT_ID);
+    assertThat(point.getAttributes().get(AttributeKey.stringKey("database_id"))).isNotNull();
 
     EasyMock.verify(rpcFactoryMock, rpcMock);
   }
@@ -279,6 +282,18 @@ public void runInTransaction_recordsGrpcStatusCodeOnFailure() {
             .orElse(null);
     assertThat(abortedPoint).isNotNull();
     assertThat(abortedPoint.getValue()).isAtLeast(1);
+
+    // Verify that latency was recorded with the failure status code
+    Optional<MetricData> latencyMetric =
+        localMetricReader.collectAllMetrics().stream()
+            .filter(m -> m.getName().equals(LATENCY_METRIC_NAME))
+            .findFirst();
+    assertThat(latencyMetric.isPresent()).isTrue();
+    HistogramPointData latencyPoint =
+        latencyMetric.get().getHistogramData().getPoints().stream().findFirst().orElse(null);
+    assertThat(latencyPoint).isNotNull();
+    assertThat(latencyPoint.getAttributes().get(AttributeKey.stringKey("status")))
+        .isEqualTo("ABORTED");
   }
 
   @Test

java-datastore/google-cloud-datastore/src/test/java/com/google/cloud/datastore/execution/AggregationQueryExecutorTest.java

@@ -37,7 +37,6 @@
 import com.google.cloud.datastore.LongValue;
 import com.google.cloud.datastore.Query;
 import com.google.cloud.datastore.spi.v1.DatastoreRpc;
-import com.google.cloud.datastore.telemetry.MetricsRecorder;
 import com.google.common.collect.ImmutableMap;
 import com.google.datastore.v1.AggregationResultBatch;
 import com.google.datastore.v1.RunAggregationQueryRequest;
@@ -67,8 +66,7 @@ public void setUp() throws Exception {
     queryExecutor =
         new AggregationQueryExecutor(
             mockRpc,
-            datastoreOptions,
-            MetricsRecorder.getInstance(datastoreOptions.getOpenTelemetryOptions()));
+            datastoreOptions);
   }
 
   @Test

java-datastore/google-cloud-datastore/src/test/java/com/google/cloud/datastore/telemetry/OpenTelemetryMetricsRecorderTest.java

@@ -69,7 +69,7 @@ public void recordTransactionLatency_recordsHistogramWithAttributes() {
 
     assertThat(latencyMetric).isNotNull();
     assertThat(latencyMetric.getDescription())
-        .isEqualTo("Total latency for successful transaction operations");
+        .isEqualTo("Total latency of transaction operations");
     assertThat(latencyMetric.getUnit()).isEqualTo("ms");
 
     HistogramPointData point =