Repository: bazelbuild/bazel
Issue: [6.3.0] Ensure outputs are downloaded before sending TargetComplete event via BEP

Forked from #17798

Issue comments:
Fixed by #18815.

Pull request review comments:

Patch that resolved the issue:
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceModule.java b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceModule.java
index 4de92bf74c588d..b99dceead675c4 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceModule.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceModule.java
@@ -38,6 +38,7 @@
 import com.google.devtools.build.lib.buildeventservice.client.BuildEventServiceClient;
 import com.google.devtools.build.lib.buildeventstream.AnnounceBuildEventTransportsEvent;
 import com.google.devtools.build.lib.buildeventstream.BuildEventArtifactUploader;
+import com.google.devtools.build.lib.buildeventstream.BuildEventLocalFileSynchronizer;
 import com.google.devtools.build.lib.buildeventstream.BuildEventProtocolOptions;
 import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos.Aborted.AbortReason;
 import com.google.devtools.build.lib.buildeventstream.BuildEventTransport;
@@ -90,6 +91,7 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 
 /**
@@ -753,6 +755,19 @@ private ImmutableSet<BuildEventTransport> createBepTransports(
       CountingArtifactGroupNamer artifactGroupNamer)
       throws IOException {
     ImmutableSet.Builder<BuildEventTransport> bepTransportsBuilder = new ImmutableSet.Builder<>();
+    BuildEventLocalFileSynchronizer synchronizer =
+        localFiles -> {
+          var outputService = cmdEnv.getOutputService();
+          if (outputService == null || localFiles.isEmpty()) {
+            return Futures.immediateVoidFuture();
+          }
+
+          var files =
+              localFiles.stream()
+                  .map(localFile -> localFile.path.asFragment())
+                  .collect(Collectors.toList());
+          return outputService.waitOutputDownloads(files);
+        };
 
     if (!Strings.isNullOrEmpty(besStreamOptions.buildEventTextFile)) {
       try {
@@ -766,7 +781,11 @@ private ImmutableSet<BuildEventTransport> createBepTransports(
                 : new LocalFilesArtifactUploader();
         bepTransportsBuilder.add(
             new TextFormatFileTransport(
-                bepTextOutputStream, bepOptions, localFileUploader, artifactGroupNamer));
+                bepTextOutputStream,
+                bepOptions,
+                localFileUploader,
+                artifactGroupNamer,
+                synchronizer));
       } catch (IOException exception) {
         // TODO(b/125216340): Consider making this a warning instead of an error once the
         //  associated bug has been resolved.
@@ -793,7 +812,11 @@ private ImmutableSet<BuildEventTransport> createBepTransports(
                 : new LocalFilesArtifactUploader();
         bepTransportsBuilder.add(
             new BinaryFormatFileTransport(
-                bepBinaryOutputStream, bepOptions, localFileUploader, artifactGroupNamer));
+                bepBinaryOutputStream,
+                bepOptions,
+                localFileUploader,
+                artifactGroupNamer,
+                synchronizer));
       } catch (IOException exception) {
         // TODO(b/125216340): Consider making this a warning instead of an error once the
         //  associated bug has been resolved.
@@ -819,7 +842,11 @@ private ImmutableSet<BuildEventTransport> createBepTransports(
                 : new LocalFilesArtifactUploader();
         bepTransportsBuilder.add(
             new JsonFormatFileTransport(
-                bepJsonOutputStream, bepOptions, localFileUploader, artifactGroupNamer));
+                bepJsonOutputStream,
+                bepOptions,
+                localFileUploader,
+                artifactGroupNamer,
+                synchronizer));
       } catch (IOException exception) {
         // TODO(b/125216340): Consider making this a warning instead of an error once the
         //  associated bug has been resolved.
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/BuildEventLocalFileSynchronizer.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/BuildEventLocalFileSynchronizer.java
new file mode 100644
index 00000000000000..9b693f7cf68dda
--- /dev/null
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/BuildEventLocalFileSynchronizer.java
@@ -0,0 +1,29 @@
+// Copyright 2023 The Bazel Authors. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.google.devtools.build.lib.buildeventstream;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.devtools.build.lib.buildeventstream.BuildEvent.LocalFile;
+import java.util.Collection;
+
+/**
+ * An interface that is used to wait for downloads of remote outputs before sending out local BEP
+ * events that reference these outputs.
+ */
+public interface BuildEventLocalFileSynchronizer {
+  BuildEventLocalFileSynchronizer NO_OP = localFiles -> Futures.immediateVoidFuture();
+
+  ListenableFuture<Void> waitForLocalFileDownloads(Collection<LocalFile> localFiles);
+}
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransport.java
index c3a83c0a1f8ea8..d1f71a84f381e1 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransport.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransport.java
@@ -17,6 +17,7 @@
 import com.google.devtools.build.lib.buildeventstream.ArtifactGroupNamer;
 import com.google.devtools.build.lib.buildeventstream.BuildEvent;
 import com.google.devtools.build.lib.buildeventstream.BuildEventArtifactUploader;
+import com.google.devtools.build.lib.buildeventstream.BuildEventLocalFileSynchronizer;
 import com.google.devtools.build.lib.buildeventstream.BuildEventProtocolOptions;
 import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos;
 import com.google.devtools.build.lib.buildeventstream.BuildEventTransport;
@@ -34,8 +35,9 @@ public BinaryFormatFileTransport(
       BufferedOutputStream outputStream,
       BuildEventProtocolOptions options,
       BuildEventArtifactUploader uploader,
-      ArtifactGroupNamer namer) {
-    super(outputStream, options, uploader, namer);
+      ArtifactGroupNamer namer,
+      BuildEventLocalFileSynchronizer synchronizer) {
+    super(outputStream, options, uploader, namer, synchronizer);
   }
 
   @Override
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java
index 7bc427d7849342..d1473e7ba53cbe 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java
@@ -28,6 +28,7 @@
 import com.google.devtools.build.lib.buildeventstream.BuildEvent;
 import com.google.devtools.build.lib.buildeventstream.BuildEventArtifactUploader;
 import com.google.devtools.build.lib.buildeventstream.BuildEventContext;
+import com.google.devtools.build.lib.buildeventstream.BuildEventLocalFileSynchronizer;
 import com.google.devtools.build.lib.buildeventstream.BuildEventProtocolOptions;
 import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos;
 import com.google.devtools.build.lib.buildeventstream.BuildEventTransport;
@@ -66,6 +67,7 @@ abstract class FileTransport implements BuildEventTransport {
   private final BuildEventArtifactUploader uploader;
   private final SequentialWriter writer;
   private final ArtifactGroupNamer namer;
+  private final BuildEventLocalFileSynchronizer synchronizer;
 
   private final ScheduledExecutorService timeoutExecutor =
       MoreExecutors.listeningDecorator(
@@ -76,12 +78,14 @@ abstract class FileTransport implements BuildEventTransport {
       BufferedOutputStream outputStream,
       BuildEventProtocolOptions options,
       BuildEventArtifactUploader uploader,
-      ArtifactGroupNamer namer) {
+      ArtifactGroupNamer namer,
+      BuildEventLocalFileSynchronizer synchronizer) {
     this.uploader = uploader;
     this.options = options;
     this.writer =
         new SequentialWriter(outputStream, this::serializeEvent, uploader, timeoutExecutor);
     this.namer = namer;
+    this.synchronizer = synchronizer;
   }
 
   @ThreadSafe
@@ -280,12 +284,14 @@ private ListenableFuture<BuildEventStreamProtos.BuildEvent> asStreamProto(
       BuildEvent event, ArtifactGroupNamer namer) {
     checkNotNull(event);
 
+    var localFiles = event.referencedLocalFiles();
+    ListenableFuture<?> localFileDownloads = synchronizer.waitForLocalFileDownloads(localFiles);
     ListenableFuture<PathConverter> converterFuture =
-        uploader.uploadReferencedLocalFiles(event.referencedLocalFiles());
+        uploader.uploadReferencedLocalFiles(localFiles);
     ListenableFuture<?> remoteUploads =
         uploader.waitForRemoteUploads(event.remoteUploads(), timeoutExecutor);
     return Futures.transform(
-        Futures.allAsList(converterFuture, remoteUploads),
+        Futures.allAsList(localFileDownloads, converterFuture, remoteUploads),
         results -> {
           BuildEventContext context =
               new BuildEventContext() {
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/JsonFormatFileTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/JsonFormatFileTransport.java
index a25074164cba7a..dc6bcd9e017e2c 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/JsonFormatFileTransport.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/JsonFormatFileTransport.java
@@ -18,6 +18,7 @@
 
 import com.google.devtools.build.lib.buildeventstream.ArtifactGroupNamer;
 import com.google.devtools.build.lib.buildeventstream.BuildEventArtifactUploader;
+import com.google.devtools.build.lib.buildeventstream.BuildEventLocalFileSynchronizer;
 import com.google.devtools.build.lib.buildeventstream.BuildEventProtocolOptions;
 import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos;
 import com.google.devtools.build.lib.buildeventstream.BuildEventTransport;
@@ -34,8 +35,9 @@ public JsonFormatFileTransport(
       BufferedOutputStream outputStream,
       BuildEventProtocolOptions options,
       BuildEventArtifactUploader uploader,
-      ArtifactGroupNamer namer) {
-    super(outputStream, options, uploader, namer);
+      ArtifactGroupNamer namer,
+      BuildEventLocalFileSynchronizer synchronizer) {
+    super(outputStream, options, uploader, namer, synchronizer);
   }
 
   @Override
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransport.java
index 34e26e67c1c33b..f0b88651b7fa1e 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransport.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransport.java
@@ -18,6 +18,7 @@
 
 import com.google.devtools.build.lib.buildeventstream.ArtifactGroupNamer;
 import com.google.devtools.build.lib.buildeventstream.BuildEventArtifactUploader;
+import com.google.devtools.build.lib.buildeventstream.BuildEventLocalFileSynchronizer;
 import com.google.devtools.build.lib.buildeventstream.BuildEventProtocolOptions;
 import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos;
 import com.google.devtools.build.lib.buildeventstream.BuildEventTransport;
@@ -35,8 +36,9 @@ public TextFormatFileTransport(
       BufferedOutputStream outputStream,
       BuildEventProtocolOptions options,
       BuildEventArtifactUploader uploader,
-      ArtifactGroupNamer namer) {
-    super(outputStream, options, uploader, namer);
+      ArtifactGroupNamer namer,
+      BuildEventLocalFileSynchronizer synchronizer) {
+    super(outputStream, options, uploader, namer, synchronizer);
   }
 
   @Override
diff --git a/src/main/java/com/google/devtools/build/lib/remote/AbstractActionInputPrefetcher.java b/src/main/java/com/google/devtools/build/lib/remote/AbstractActionInputPrefetcher.java
index a037811a8e70e2..02dae8d8ca4d9a 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/AbstractActionInputPrefetcher.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/AbstractActionInputPrefetcher.java
@@ -48,6 +48,7 @@
 import com.google.devtools.build.lib.events.Reporter;
 import com.google.devtools.build.lib.remote.common.CacheNotFoundException;
 import com.google.devtools.build.lib.remote.util.AsyncTaskCache;
+import com.google.devtools.build.lib.remote.util.RxFutures;
 import com.google.devtools.build.lib.remote.util.RxUtils.TransferResult;
 import com.google.devtools.build.lib.remote.util.TempPathGenerator;
 import com.google.devtools.build.lib.vfs.FileSystemUtils;
@@ -60,6 +61,7 @@
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Deque;
 import java.util.List;
 import java.util.Set;
@@ -67,6 +69,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 
 /**
@@ -760,6 +763,12 @@ public void flushOutputTree() throws InterruptedException {
     downloadCache.awaitInProgressTasks();
   }
 
+  public ListenableFuture<Void> waitDownloads(Collection<PathFragment> files) {
+    var convertedFiles = files.stream().map(file -> execRoot.getFileSystem().getPath(file)).collect(
+        Collectors.toList());
+    return RxFutures.toListenableFuture(downloadCache.waitInProgressTasks(convertedFiles));
+  }
+
   public ImmutableSet<ActionInput> getMissingActionInputs() {
     return ImmutableSet.copyOf(missingActionInputs);
   }
diff --git a/src/main/java/com/google/devtools/build/lib/remote/RemoteOutputService.java b/src/main/java/com/google/devtools/build/lib/remote/RemoteOutputService.java
index 7cf03eb399862f..c1b9a6d73b465e 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/RemoteOutputService.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/RemoteOutputService.java
@@ -19,6 +19,8 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.eventbus.Subscribe;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
 import com.google.devtools.build.lib.actions.Action;
 import com.google.devtools.build.lib.actions.ActionExecutionMetadata;
 import com.google.devtools.build.lib.actions.ActionInputMap;
@@ -34,10 +36,12 @@
 import com.google.devtools.build.lib.vfs.FileSystem;
 import com.google.devtools.build.lib.vfs.ModifiedFileSet;
 import com.google.devtools.build.lib.vfs.OutputService;
+import com.google.devtools.build.lib.vfs.Path;
 import com.google.devtools.build.lib.vfs.PathFragment;
 import com.google.devtools.build.lib.vfs.Root;
 import com.google.devtools.build.skyframe.SkyFunction.Environment;
 import java.io.IOException;
+import java.util.Collection;
 import java.util.Map;
 import java.util.UUID;
 import javax.annotation.Nullable;
@@ -111,6 +115,13 @@ public void flushOutputTree() throws InterruptedException {
     }
   }
 
+  public ListenableFuture<Void> waitOutputDownloads(Collection<PathFragment> files) {
+    if (actionInputFetcher != null) {
+      return actionInputFetcher.waitDownloads(files);
+    }
+    return Futures.immediateVoidFuture();
+  }
+
   @Override
   public void finalizeBuild(boolean buildSuccessful) {
     // Intentionally left empty.
diff --git a/src/main/java/com/google/devtools/build/lib/remote/util/AsyncTaskCache.java b/src/main/java/com/google/devtools/build/lib/remote/util/AsyncTaskCache.java
index cd91a72b503226..5d63d2927053c2 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/util/AsyncTaskCache.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/util/AsyncTaskCache.java
@@ -28,6 +28,7 @@
 import io.reactivex.rxjava3.functions.Action;
 import io.reactivex.rxjava3.subjects.AsyncSubject;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -357,6 +358,32 @@ public void shutdown() {
     }
   }
 
+  /**
+   * Returns a {@link Completable} which will completes once the in-progress tasks identified by
+   * {@code keys} are completed. Tasks submitted after the call are not waited.
+   */
+  public Completable waitInProgressTasks(Collection<KeyT> keys) {
+    return Completable.defer(
+        () -> {
+          List<Execution> executions = new ArrayList<>();
+          synchronized (lock) {
+            for (var key : keys) {
+              var execution = inProgress.get(key);
+              if (execution != null) {
+                executions.add(execution);
+              }
+            }
+          }
+          if (executions.isEmpty()) {
+            return Completable.complete();
+          }
+
+          return Completable.fromPublisher(
+              Flowable.fromIterable(executions)
+                  .flatMapSingle(e -> Single.fromObservable(e.completion)));
+        });
+  }
+
   /**
    * Waits for the in-progress tasks to finish. Any tasks that are submitted after the call are not
    * waited.
@@ -535,6 +562,14 @@ public void shutdown() {
       cache.shutdown();
     }
 
+    /**
+     * Returns a {@link Completable} which will completes once the in-progress tasks identified by
+     * {@code keys} are completed. Tasks submitted after the call are not waited.
+     */
+    public Completable waitInProgressTasks(Collection<KeyT> keys) {
+      return cache.waitInProgressTasks(keys);
+    }
+
     /**
      * Waits for the in-progress tasks to finish. Any tasks that are submitted after the call are
      * not waited.
diff --git a/src/main/java/com/google/devtools/build/lib/vfs/OutputService.java b/src/main/java/com/google/devtools/build/lib/vfs/OutputService.java
index 45ce3d779ea40f..3fb78d67eb008f 100644
--- a/src/main/java/com/google/devtools/build/lib/vfs/OutputService.java
+++ b/src/main/java/com/google/devtools/build/lib/vfs/OutputService.java
@@ -17,6 +17,8 @@
 import com.google.common.collect.ImmutableCollection;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
 import com.google.devtools.build.lib.actions.Action;
 import com.google.devtools.build.lib.actions.ActionExecutionMetadata;
 import com.google.devtools.build.lib.actions.ActionInputMap;
@@ -33,6 +35,7 @@
 import com.google.devtools.build.lib.util.AbruptExitException;
 import com.google.devtools.build.skyframe.SkyFunction.Environment;
 import java.io.IOException;
+import java.util.Collection;
 import java.util.Map;
 import java.util.UUID;
 import javax.annotation.Nullable;
@@ -121,6 +124,10 @@ ModifiedFileSet startBuild(EventHandler eventHandler, UUID buildId, boolean fina
   /** Flush and wait for in-progress downloads. */
   default void flushOutputTree() throws InterruptedException {}
 
+  default ListenableFuture<Void> waitOutputDownloads(Collection<PathFragment> files) {
+    return Futures.immediateVoidFuture();
+  }
+
   /**
    * Finish the build.
    *


Test patch:
diff --git a/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransportTest.java b/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransportTest.java
index 527bccebc526ea..5286b50e0cb146 100644
--- a/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransportTest.java
+++ b/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransportTest.java
@@ -32,6 +32,7 @@
 import com.google.devtools.build.lib.buildeventstream.BuildEventArtifactUploader;
 import com.google.devtools.build.lib.buildeventstream.BuildEventContext;
 import com.google.devtools.build.lib.buildeventstream.BuildEventIdUtil;
+import com.google.devtools.build.lib.buildeventstream.BuildEventLocalFileSynchronizer;
 import com.google.devtools.build.lib.buildeventstream.BuildEventProtocolOptions;
 import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos;
 import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos.BuildEventId;
@@ -107,7 +108,8 @@ public void testCreatesFileAndWritesProtoBinaryFormat() throws Exception {
             outputStream,
             defaultOpts,
             new LocalFilesArtifactUploader(),
-            artifactGroupNamer);
+            artifactGroupNamer,
+            BuildEventLocalFileSynchronizer.NO_OP);
     transport.sendBuildEvent(buildEvent);
 
     BuildEventStreamProtos.BuildEvent progress =
@@ -157,7 +159,12 @@ public boolean mayBeSlow() {
     BufferedOutputStream outputStream =
         new BufferedOutputStream(Files.newOutputStream(Paths.get(output.getAbsolutePath())));
     BinaryFormatFileTransport transport =
-        new BinaryFormatFileTransport(outputStream, defaultOpts, uploader, artifactGroupNamer);
+        new BinaryFormatFileTransport(
+            outputStream,
+            defaultOpts,
+            uploader,
+            artifactGroupNamer,
+            BuildEventLocalFileSynchronizer.NO_OP);
     transport.sendBuildEvent(event1);
 
     ExecutionException expected =
@@ -189,7 +196,8 @@ public void testWriteWhenFileClosed() throws Exception {
             outputStream,
             defaultOpts,
             new LocalFilesArtifactUploader(),
-            artifactGroupNamer);
+            artifactGroupNamer,
+            BuildEventLocalFileSynchronizer.NO_OP);
 
     transport.close().get();
 
@@ -220,7 +228,8 @@ public void testWriteWhenTransportClosed() throws Exception {
             outputStream,
             defaultOpts,
             new LocalFilesArtifactUploader(),
-            artifactGroupNamer);
+            artifactGroupNamer,
+            BuildEventLocalFileSynchronizer.NO_OP);
 
     transport.sendBuildEvent(buildEvent);
     Future<Void> closeFuture = transport.close();
@@ -267,7 +276,12 @@ public boolean mayBeSlow() {
     BufferedOutputStream outputStream =
         new BufferedOutputStream(Files.newOutputStream(Paths.get(output.getAbsolutePath())));
     BinaryFormatFileTransport transport =
-        new BinaryFormatFileTransport(outputStream, defaultOpts, uploader, artifactGroupNamer);
+        new BinaryFormatFileTransport(
+            outputStream,
+            defaultOpts,
+            uploader,
+            artifactGroupNamer,
+            BuildEventLocalFileSynchronizer.NO_OP);
     transport.sendBuildEvent(event1);
     transport.sendBuildEvent(event2);
     transport.close().get();
@@ -307,7 +321,12 @@ public boolean mayBeSlow() {
     BufferedOutputStream outputStream =
         new BufferedOutputStream(Files.newOutputStream(Paths.get(output.getAbsolutePath())));
     BinaryFormatFileTransport transport =
-        new BinaryFormatFileTransport(outputStream, defaultOpts, uploader, artifactGroupNamer);
+        new BinaryFormatFileTransport(
+            outputStream,
+            defaultOpts,
+            uploader,
+            artifactGroupNamer,
+            BuildEventLocalFileSynchronizer.NO_OP);
     transport.sendBuildEvent(event1);
     transport.close().get();
 
@@ -345,7 +364,12 @@ public boolean mayBeSlow() {
     BufferedOutputStream outputStream =
         new BufferedOutputStream(Files.newOutputStream(Paths.get(output.getAbsolutePath())));
     BinaryFormatFileTransport transport =
-        new BinaryFormatFileTransport(outputStream, defaultOpts, uploader, artifactGroupNamer);
+        new BinaryFormatFileTransport(
+            outputStream,
+            defaultOpts,
+            uploader,
+            artifactGroupNamer,
+            BuildEventLocalFileSynchronizer.NO_OP);
     transport.sendBuildEvent(event);
     ListenableFuture<Void> closeFuture = transport.close();
 
diff --git a/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/JsonFormatFileTransportTest.java b/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/JsonFormatFileTransportTest.java
index 400bdff6d32e68..9229df317f3460 100644
--- a/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/JsonFormatFileTransportTest.java
+++ b/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/JsonFormatFileTransportTest.java
@@ -20,6 +20,7 @@
 import com.google.devtools.build.lib.buildeventstream.ArtifactGroupNamer;
 import com.google.devtools.build.lib.buildeventstream.BuildEvent;
 import com.google.devtools.build.lib.buildeventstream.BuildEventContext;
+import com.google.devtools.build.lib.buildeventstream.BuildEventLocalFileSynchronizer;
 import com.google.devtools.build.lib.buildeventstream.BuildEventProtocolOptions;
 import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos;
 import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos.BuildStarted;
@@ -87,7 +88,8 @@ public void testCreatesFileAndWritesProtoJsonFormat() throws Exception {
             outputStream,
             defaultOpts,
             new LocalFilesArtifactUploader(),
-            artifactGroupNamer);
+            artifactGroupNamer,
+            BuildEventLocalFileSynchronizer.NO_OP);
     transport.sendBuildEvent(buildEvent);
 
     transport.close().get();
@@ -155,7 +157,11 @@ public void testFlushesStreamAfterSmallWrites() throws Exception {
 
     JsonFormatFileTransport transport =
         new JsonFormatFileTransport(
-            wrappedOutputStream, defaultOpts, new LocalFilesArtifactUploader(), artifactGroupNamer);
+            wrappedOutputStream,
+            defaultOpts,
+            new LocalFilesArtifactUploader(),
+            artifactGroupNamer,
+            BuildEventLocalFileSynchronizer.NO_OP);
 
     transport.sendBuildEvent(buildEvent);
     Thread.sleep(transport.getFlushInterval().toMillis() * 3);
diff --git a/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransportTest.java b/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransportTest.java
index a307a129ebc92e..2511852aa8bf6c 100644
--- a/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransportTest.java
+++ b/src/test/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransportTest.java
@@ -22,6 +22,7 @@
 import com.google.devtools.build.lib.buildeventstream.ArtifactGroupNamer;
 import com.google.devtools.build.lib.buildeventstream.BuildEvent;
 import com.google.devtools.build.lib.buildeventstream.BuildEventContext;
+import com.google.devtools.build.lib.buildeventstream.BuildEventLocalFileSynchronizer;
 import com.google.devtools.build.lib.buildeventstream.BuildEventProtocolOptions;
 import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos;
 import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos.BuildStarted;
@@ -88,7 +89,8 @@ public void testCreatesFileAndWritesProtoTextFormat() throws Exception {
             outputStream,
             defaultOpts,
             new LocalFilesArtifactUploader(),
-            artifactGroupNamer);
+            artifactGroupNamer,
+            BuildEventLocalFileSynchronizer.NO_OP);
     transport.sendBuildEvent(buildEvent);
 
     BuildEventStreamProtos.BuildEvent progress =