BES/BEP interrupt audit:

- don't leak BES uploading between builds on interrupt
- don't crash on timeout

RELNOTES: None
PiperOrigin-RevId: 229929479

src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceModule.java

@@ -44,6 +44,7 @@
 import com.google.devtools.build.lib.runtime.SynchronizedOutputStream;
 import com.google.devtools.build.lib.util.AbruptExitException;
 import com.google.devtools.build.lib.util.ExitCode;
+import com.google.devtools.build.lib.util.LoggingUtil;
 import com.google.devtools.build.lib.util.io.OutErr;
 import com.google.devtools.common.options.OptionsBase;
 import com.google.devtools.common.options.OptionsParsingException;
@@ -142,10 +143,14 @@ public void blazeShutdownOnCrash() {
   @Override
   public void afterCommand() {
     if (streamer != null) {
-      // This should not occur, but close with an internal error if a {@link BuildEventStreamer} bug
-      // manifests as an unclosed streamer.
-      logger.warning("Attempting to close BES streamer after command");
-      streamer.close(AbortReason.INTERNAL);
+      if (!streamer.isClosed()) {
+        // This should not occur, but close with an internal error if a {@link BuildEventStreamer}
+        // bug manifests as an unclosed streamer.
+        logger.warning("Attempting to close BES streamer after command");
+        String msg = "BES was not properly closed";
+        LoggingUtil.logToRemote(Level.WARNING, msg, new IllegalStateException(msg));
+        streamer.close(AbortReason.INTERNAL);
+      }
       this.streamer = null;
     }
 

src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceTransport.java

@@ -15,6 +15,7 @@
 package com.google.devtools.build.lib.buildeventservice;
 
 import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.util.concurrent.Uninterruptibles.getUninterruptibly;
 import static com.google.devtools.build.v1.BuildStatus.Result.COMMAND_FAILED;
 import static com.google.devtools.build.v1.BuildStatus.Result.COMMAND_SUCCEEDED;
 import static com.google.devtools.build.v1.BuildStatus.Result.UNKNOWN_STATUS;
@@ -352,7 +353,7 @@ public ListenableFuture<Void> close() {
     }
 
     /** Stops the upload immediately. Enqueued events that have not been sent yet will be lost. */
-    public void closeOnTimeout() {
+    void closeOnTimeout() {
       synchronized (lock) {
         if (uploadThread != null) {
           if (uploadThread.isInterrupted()) {
@@ -397,9 +398,9 @@ public void run() {
         try {
           logInfo(e, "Aborting the BES upload due to having received an interrupt");
           synchronized (lock) {
-            if (interruptCausedByTimeout) {
-              exitFunc.accept("The Build Event Protocol upload timed out", e);
-            }
+            Preconditions.checkState(
+                interruptCausedByTimeout, "Unexpected interrupt on BES uploader thread");
+            exitFunc.accept("The Build Event Protocol upload timed out", e);
           }
         } finally {
           // TODO(buchgr): Due to b/113035235 exitFunc needs to be called before the close future
@@ -677,8 +678,8 @@ private void startCloseTimer(ListenableFuture<Void> closeFuture, Duration closeT
               () -> {
                 // Call closeOnTimeout() if the future does not complete within closeTimeout
                 try {
-                  closeFuture.get(closeTimeout.toMillis(), TimeUnit.MILLISECONDS);
-                } catch (InterruptedException | TimeoutException e) {
+                  getUninterruptibly(closeFuture, closeTimeout.toMillis(), TimeUnit.MILLISECONDS);
+                } catch (TimeoutException e) {
                   closeOnTimeout();
                 } catch (ExecutionException e) {
                   if (e.getCause() instanceof TimeoutException) {

src/main/java/com/google/devtools/build/lib/runtime/BuildEventStreamer.java

@@ -30,6 +30,7 @@
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.common.util.concurrent.Uninterruptibles;
 import com.google.devtools.build.lib.actions.ActionExecutedEvent;
 import com.google.devtools.build.lib.actions.Artifact;
 import com.google.devtools.build.lib.actions.ArtifactPathResolver;
@@ -67,6 +68,7 @@
 import com.google.devtools.build.lib.events.EventHandler;
 import com.google.devtools.build.lib.events.Reporter;
 import com.google.devtools.build.lib.pkgcache.TargetParsingCompleteEvent;
+import com.google.devtools.build.lib.util.LoggingUtil;
 import com.google.devtools.build.lib.util.Pair;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -75,10 +77,12 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
@@ -408,11 +412,12 @@ public void close(@Nullable AbortReason reason) {
         }
 
         ScheduledFuture<?> f = bepUploadWaitEvent(executor);
-        // Wait for all transports to close.
-        Futures.allAsList(closeFutures).get();
+        // Wait for all transports to close, ignoring interrupts.
+        Uninterruptibles.getUninterruptibly(Futures.allAsList(closeFutures));
         f.cancel(true);
-      } catch (Exception e) {
-        logger.severe("Failed to close a build event transport: " + e);
+      } catch (ExecutionException e) {
+        logger.log(Level.SEVERE, "Failed to close a build event transport", e);
+        LoggingUtil.logToRemote(Level.SEVERE, "Failed to close a build event transport", e);
       }
     } finally {
       if (executor != null) {