Properly wire up BES half-close. The lack thereof was a simple oversi…

…ght.

With this change, --bes_upload_mode=fully_async works properly.

RELNOTES: None
PiperOrigin-RevId: 251295660

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

@@ -67,6 +67,11 @@ public ListenableFuture<Void> close() {
     return besUploader.close();
   }
 
+  @Override
+  public ListenableFuture<Void> getHalfCloseFuture() {
+    return besUploader.getHalfCloseFuture();
+  }
+
   @Override
   public BuildEventArtifactUploader getUploader() {
     return besUploader.getLocalFileUploader();

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

@@ -154,6 +154,10 @@ private BuildEventServiceUploader(
     this.clock = clock;
     this.namer = namer;
     this.eventBus = eventBus;
+    // Ensure the half-close future is closed once the upload is complete. This is usually a no-op,
+    // but makes sure we half-close in case of error / interrupt.
+    closeFuture.addListener(
+        () -> halfCloseFuture.setFuture(closeFuture), MoreExecutors.directExecutor());
   }
 
   BuildEventArtifactUploader getLocalFileUploader() {
@@ -255,6 +259,10 @@ private void closeNow() {
     }
   }
 
+  ListenableFuture<Void> getHalfCloseFuture() {
+    return halfCloseFuture;
+  }
+
   private void logAndExitAbruptly(String message, ExitCode exitCode, Throwable cause) {
     checkState(!exitCode.equals(ExitCode.SUCCESS));
     logger.severe(message);
@@ -422,6 +430,7 @@ private void publishBuildEvents()
               streamContext.sendOverStream(request);
               streamContext.halfCloseStream();
               halfCloseFuture.set(null);
+              logger.info("BES uploader is half-closed");
             }
             break;