Remediate some issues to successfully pass bazel tests

Author: chenj-hub

src/main/java/build/buildfarm/cas/cfc/CASFileCache.java

@@ -3209,6 +3209,7 @@ void commit() throws IOException {
         Entry existingEntry = null;
         boolean inserted = false;
         try {
+          // acquire the key lock
           log.log(Level.FINEST, "comitting " + key + " from " + writePath);
           Path cachePath = CASFileCache.this.getPath(key);
           CASFileCache.this.renamePath(writePath, cachePath);

src/main/java/build/buildfarm/common/redis/RedisClient.java

@@ -83,7 +83,7 @@ public JedisMisconfigurationException(final String message, final Throwable caus
   // We store the factory in case we want to re-create the jedis client.
   private Supplier<UnifiedJedis> unifiedJedisFactory;
 
-  private final UnifiedJedis jedis;
+  private UnifiedJedis jedis;
 
   private boolean closed = false;
 

src/main/java/build/buildfarm/instance/shard/RedisShardBackplane.java

@@ -538,18 +538,12 @@ public void start(String clientPublicName) throws IOException {
     // Construct a single redis client to be used throughout the entire backplane.
     // We wish to avoid various synchronous and error handling issues that could occur when using
     // multiple clients.
-    client =
-        new RedisClient(
-            jedisClusterFactory,
-            configs.getBackplane().getReconnectClientAttempts(),
-            configs.getBackplane().getReconnectClientWaitDurationMs());
-  // Create containers that make up the backplane
     start(new RedisClient(jedisClusterFactory.get()), clientPublicName);
-}
+  }
 
   private void start(RedisClient client, String clientPublicName) throws IOException {
-      // Create containers that make up the backplane
-      start(client, DistributedStateCreator.create(client), clientPublicName);
+    // Create containers that make up the backplane
+    start(client, DistributedStateCreator.create(client), clientPublicName);
   }
 
   @VisibleForTesting

src/main/java/build/buildfarm/instance/shard/ServerInstance.java

@@ -247,7 +247,7 @@ public class ServerInstance extends NodeInstance {
   private Cache<RequestMetadata, Boolean> recentCacheServedExecutions;
 
   private final Random rand = new Random();
-  private final Writes writes;
+  private final Writes writes = new Writes(this::writeInstanceSupplier);
   private final int maxCpu;
   private final int maxRequeueAttempts;
 
@@ -382,7 +382,6 @@ public ServerInstance(
     this.actionCacheFetchService = actionCacheFetchService;
     backplane.setOnUnsubscribe(this::stop);
 
-    this.writes = new Writes(writeInstanceCacheLoader());
     initializeCaches();
 
     remoteInputStreamFactory =
@@ -1248,35 +1247,9 @@ public void onSuccess(List<String> workersList) {
     protected abstract void onQueue(Deque<String> workers);
   }
 
-  private CacheLoader<BlobWriteKey, Instance> writeInstanceCacheLoader() {
-    return new CacheLoader<BlobWriteKey, Instance>() {
-      @SuppressWarnings("NullableProblems")
-      @Override
-      public Instance load(BlobWriteKey key) {
-        String instance = null;
-        // Per the REAPI the identifier should end up as a unique UUID per a
-        // client level - adding bytes to further mitigate collisions and not
-        // store the entire BlobWriteKey.
-        String blobKey = key.getIdentifier() + "." + key.getDigest().getSizeBytes();
-        try {
-          instance = backplane.getWriteInstance(blobKey);
-          if (instance != null) {
-            return workerStub(instance);
-          }
-        } catch (IOException e) {
-          log.log(Level.WARNING, "error getting write instance for " + instance, e);
-        }
-
-        instance = getRandomWorker();
-        try {
-          backplane.setWriteInstance(blobKey, instance);
-          log.log(Level.INFO, "set write-instance: " + blobKey + " -> " + instance); // TODO: [jmarino]: remove
-        } catch (IOException e) {
-          log.log(Level.WARNING, "error getting write instance for " + instance, e);
-        }
-        return workerStub(instance);
-      }
-    };
+  private Instance writeInstanceSupplier() {
+    String worker = getRandomWorker();
+    return workerStub(worker);
   }
 
   String getRandomWorker() {

src/main/java/build/buildfarm/worker/shard/CFCExecFileSystem.java

@@ -40,6 +40,7 @@
 import build.buildfarm.cas.ContentAddressableStorage;
 import build.buildfarm.cas.cfc.CASFileCache;
 import build.buildfarm.common.BuildfarmExecutors;
+import build.buildfarm.common.DigestUtil;
 import build.buildfarm.common.io.Directories;
 import build.buildfarm.common.io.Dirent;
 import build.buildfarm.worker.ExecDirException;

src/test/java/build/buildfarm/worker/PipelineTest.java

@@ -58,7 +58,7 @@ public void stageThreadReturnCompletesJoin() throws InterruptedException {
           public void run() {}
         },
         1);
-    pipeline.start(null);
+    pipeline.start();
     pipeline.join();
   }
 
@@ -73,7 +73,7 @@ public void run() {
           }
         },
         1);
-    pipeline.start(null);
+    pipeline.start();
     pipeline.join();
   }