fix: fix race identified by tsan (#2891)

mutianf · web-flow · commit 0a2bf6c6bcaf · 2026-04-16T08:53:51.000-04:00
* fix: fix race identified by tsan

Change-Id: If07161b71d4b4b22f384470068562406fcb15ae5

* fix

Change-Id: Ib80ec6a3b05a779815042576a22a834bc12bd6c6
diff --git a/google-cloud-bigtable-emulator-core/src/main/java/com/google/cloud/bigtable/emulator/core/EmulatorController.java b/google-cloud-bigtable-emulator-core/src/main/java/com/google/cloud/bigtable/emulator/core/EmulatorController.java
@@ -129,8 +129,8 @@ public synchronized void start(int port)
         throw e;
       }
     }
-    Thread stdoutThread = pipeStreamToLog(process.getInputStream(), Level.INFO);
-    Thread stderrThread = pipeStreamToLog(process.getErrorStream(), Level.WARNING);
+    stdoutThread = pipeStreamToLog(process.getInputStream(), Level.INFO);
+    stderrThread = pipeStreamToLog(process.getErrorStream(), Level.WARNING);
     isStopped = false;
 
     shutdownHook =
diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/PreparedStatementImpl.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/PreparedStatementImpl.java
@@ -154,7 +154,7 @@ ApiFuture<PrepareResponse> getFreshPlan() {
    * Check the expiry of the current plan, if it's future is resolved. If we are within 1s of
    * expiry, call startBackgroundRefresh with the version of the latest PrepareQuery.
    */
-  void backgroundRefreshIfNeeded() {
+  synchronized void backgroundRefreshIfNeeded() {
     PrepareQueryState localState = this.currentState.get();
     if (localState.maybeBackgroundRefresh().isPresent()) {
       // We already have an ongoing refresh
@@ -183,7 +183,7 @@ void backgroundRefreshIfNeeded() {
    * Returns the most recently refreshed PreparedQueryData. It may still be refreshing if the
    * previous plan has expired.
    */
-  public PreparedQueryData getLatestPrepareResponse() {
+  public synchronized PreparedQueryData getLatestPrepareResponse() {
     PrepareQueryState localState = currentState.get();
     if (localState.maybeBackgroundRefresh().isPresent()
         && localState.maybeBackgroundRefresh().get().prepareFuture().isDone()) {
diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/exporter/BigtableCloudMonitoringExporter.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/exporter/BigtableCloudMonitoringExporter.java
@@ -79,7 +79,7 @@ public class BigtableCloudMonitoringExporter implements MetricExporter {
   private final MetricServiceClient client;
 
   private final AtomicReference<State> state;
-  private CompletableResultCode lastExportCode;
+  private AtomicReference<CompletableResultCode> lastExportCode = new AtomicReference<>();
   private final AtomicBoolean exportFailureLogged = new AtomicBoolean(false);
 
   private enum State {
@@ -149,8 +149,9 @@ public void close() {
   public CompletableResultCode export(Collection<MetricData> metricData) {
     Preconditions.checkState(state.get() != State.Closed, "Exporter is closed");
 
-    lastExportCode = doExport(metricData);
-    return lastExportCode;
+    CompletableResultCode result = doExport(metricData);
+    lastExportCode.set(result);
+    return result;
   }
 
   private CompletableResultCode doExport(Collection<MetricData> metricData) {
@@ -194,7 +195,7 @@ public void onFailure(Throwable throwable) {
               RuntimeException asyncWrapper = new RuntimeException("export failed", throwable);
               asyncWrapper.setStackTrace(stackTrace);
 
-              if (state.get() != State.Closing && state.get() != State.Closed) {
+              if (state.get() != State.Closing || state.get() != State.Closed) {
                 // ignore the export warning when client is shutting down
                 LOGGER.log(Level.WARNING, msg, asyncWrapper);
               }
@@ -231,8 +232,8 @@ private List<ApiFuture<Empty>> exportTimeSeries(
 
   @Override
   public CompletableResultCode flush() {
-    if (lastExportCode != null) {
-      return lastExportCode;
+    if (lastExportCode.get() != null) {
+      return lastExportCode.get();
     }
     return CompletableResultCode.ofSuccess();
   }
diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/DynamicFlowControlStats.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/DynamicFlowControlStats.java
@@ -94,7 +94,7 @@ synchronized void update(long value) {
       weightedCount += weight;
     }
 
-    double getMean() {
+    synchronized double getMean() {
       return mean;
     }
 

Original file line number	Diff line number	Diff line change
`@@ -129,8 +129,8 @@ public synchronized void start(int port)`
`129`	`129`	`throw e;`
`130`	`130`	`}`
`131`	`131`	`}`
`132`		`- Thread stdoutThread = pipeStreamToLog(process.getInputStream(), Level.INFO);`
`133`		`- Thread stderrThread = pipeStreamToLog(process.getErrorStream(), Level.WARNING);`
	`132`	`+ stdoutThread = pipeStreamToLog(process.getInputStream(), Level.INFO);`
	`133`	`+ stderrThread = pipeStreamToLog(process.getErrorStream(), Level.WARNING);`
`134`	`134`	`isStopped = false;`
`135`	`135`
`136`	`136`	`shutdownHook =`
Original file line number	Diff line number	Diff line change
`@@ -94,7 +94,7 @@ synchronized void update(long value) {`
`94`	`94`	`weightedCount += weight;`
`95`	`95`	`}`
`96`	`96`
`97`		`- double getMean() {`
	`97`	`+ synchronized double getMean() {`
`98`	`98`	`return mean;`
`99`	`99`	`}`
`100`	`100`