fix: Fix race in emulator controller (#2836)

mutianf · web-flow · commit 17707be3f4bf · 2026-03-09T15:36:28.000Z
Fix TSan data race in EmulatorController by synchronizing logging threads.

The race condition occurred when a previous emulator process instance was shutting down
while a new instance was starting. The logging threads for stdout/stderr of the previous
process could still be reading from file descriptors that were subsequently reused for
the new process's pipes.

This fix ensures that:
1. Logging threads are tracked.
2. EmulatorController.stop() joins these threads after destroying the process.
3. EmulatorController.stop() waits for the process to exit completely.
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
@@ -44,6 +44,8 @@ public class EmulatorController {
 
   private final Path executable;
   private Process process;
+  private Thread stdoutThread;
+  private Thread stderrThread;
   private boolean isStopped = true;
   private Thread shutdownHook;
 
@@ -127,8 +129,8 @@ public synchronized void start(int port)
         throw e;
       }
     }
-    pipeStreamToLog(process.getInputStream(), Level.INFO);
-    pipeStreamToLog(process.getErrorStream(), Level.WARNING);
+    Thread stdoutThread = pipeStreamToLog(process.getInputStream(), Level.INFO);
+    Thread stderrThread = pipeStreamToLog(process.getErrorStream(), Level.WARNING);
     isStopped = false;
 
     shutdownHook =
@@ -164,6 +166,23 @@ public synchronized void stop() {
     } finally {
       isStopped = true;
       process.destroy();
+
+      try {
+        process.waitFor();
+        if (stdoutThread != null) {
+          stdoutThread.join();
+        }
+        if (stderrThread != null) {
+          stderrThread.join();
+        }
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        LOGGER.log(Level.WARNING, "Interrupted while waiting for emulator to stop", e);
+      } finally {
+        stdoutThread = null;
+        stderrThread = null;
+        process = null;
+      }
     }
   }
 
@@ -239,7 +258,7 @@ private static void waitForPort(int port) throws InterruptedException, TimeoutEx
   }
 
   /** Creates a thread that will pipe an {@link InputStream} to this class' Logger. */
-  private static void pipeStreamToLog(final InputStream stream, final Level level) {
+  private static Thread pipeStreamToLog(final InputStream stream, final Level level) {
     final BufferedReader reader = new BufferedReader(new InputStreamReader(stream));
 
     Thread thread =
@@ -258,6 +277,7 @@ private static void pipeStreamToLog(final InputStream stream, final Level level)
             });
     thread.setDaemon(true);
     thread.start();
+    return thread;
   }
   // </editor-fold>
 }
