Fix another race in getWorkerFor. (#18918)

This patch is covering another case, beyond the one covered in #18910.
There was still a race when multiple workers were running and all trying
to contact each other, if worker A tried to contact worker B before
worker B had been set up.

The fix is to have newWorker return null rather than throw an error in
the base class, and update getWorkerFor to retry on null.

Author: gianm

multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestWorkerClient.java

@@ -61,11 +61,13 @@ public ListenableFuture<Void> postWorkOrder(String workerTaskId, WorkOrder workO
 
   protected Worker getWorkerFor(String workerTaskId)
   {
-    final WorkerRunRef workerRunRef = inMemoryWorkers.computeIfAbsent(workerTaskId, this::newWorker);
     final Stopwatch stopwatch = Stopwatch.createStarted();
 
-    // Wait for the worker to exist
-    while (!workerRunRef.hasWorker()) {
+    // Wait for the worker to exist. It may not have been created or started up yet, especially if this is
+    // a worker trying to contact another worker.
+    WorkerRunRef workerRunRef;
+    while ((workerRunRef = inMemoryWorkers.computeIfAbsent(workerTaskId, this::newWorker)) == null
+           || !workerRunRef.hasWorker()) {
       if (stopwatch.millisElapsed() > WORKER_WAIT_TIMEOUT_MS) {
         throw new ISE(
             "Timed out after [%,d]ms waiting for worker[%s] to be registered",
@@ -88,7 +90,8 @@ protected Worker getWorkerFor(String workerTaskId)
 
   protected WorkerRunRef newWorker(String workerId)
   {
-    throw new RuntimeException("Not implemented!");
+    // Return null so getWorkerFor waits for inMemoryWorkers to be populated
+    return null;
   }
 
   @Override