Add test to verify the distributor creates sessions in parallel

Author: shs96c

java/server/test/org/openqa/selenium/grid/distributor/local/LocalDistributorTest.java

@@ -58,16 +58,27 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
+import static java.util.Collections.newSetFromMap;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
 import static org.openqa.selenium.grid.data.Availability.DRAINING;
@@ -372,6 +383,82 @@ public void testDrainNodeFromNode() {
     assertThat(localNode.isDraining()).isTrue();
   }
 
+  @Test
+  public void slowStartingNodesShouldNotCauseReservationsToBeSerialized() {
+    NewSessionQueue queue  = new LocalNewSessionQueue(
+      tracer,
+      bus,
+      new DefaultSlotMatcher(),
+      Duration.ofSeconds(2),
+      Duration.ofSeconds(2),
+      registrationSecret);
+
+    LocalDistributor distributor = new LocalDistributor(
+      tracer,
+      bus,
+      clientFactory,
+      new LocalSessionMap(tracer, bus),
+      queue,
+      new GridModel(bus),
+      new DefaultSlotSelector(),
+      registrationSecret,
+      Duration.ofMinutes(5),
+      false);
+
+    Capabilities caps = new ImmutableCapabilities("browserName", "cheese");
+
+    long delay = 4000;
+    LocalNode node = LocalNode.builder(tracer, bus, uri, uri, registrationSecret)
+      .add(caps, new TestSessionFactory(caps, (id, c) -> {
+        try {
+          Thread.sleep(delay);
+        } catch (InterruptedException e) {
+          e.printStackTrace();
+        }
+        return new Handler(c);
+      }))
+      .build();
+
+    distributor.add(node);
+
+    Set<Future<Either<SessionNotCreatedException, CreateSessionResponse>>> futures =
+      newSetFromMap(new ConcurrentHashMap<>());
+    ExecutorService service = Executors.newFixedThreadPool(2);
+
+    long start = System.currentTimeMillis();
+    futures.add(
+      service.submit(() -> distributor.newSession(new SessionRequest(
+        new RequestId(UUID.randomUUID()),
+        Instant.now(),
+        Set.of(W3C),
+        Set.of(caps),
+        Map.of(),
+        Map.of()))));
+    futures.add(
+      service.submit(() -> distributor.newSession(new SessionRequest(
+        new RequestId(UUID.randomUUID()),
+        Instant.now(),
+        Set.of(W3C),
+        Set.of(caps),
+        Map.of(),
+        Map.of()))));
+
+    futures.forEach(f -> {
+      try {
+        f.get();
+      } catch (InterruptedException e) {
+        fail("Interrupted");
+      } catch (ExecutionException e) {
+        throw new RuntimeException(e);
+      }
+    });
+
+    // If the sessions are created serially, then we expect the first
+    // session to take up to `delay` ms to complete, followed by the
+    // second session.
+    assertThat(System.currentTimeMillis() - start).isLessThan(delay * 2);
+  }
+
   private class Handler extends Session implements HttpHandler {
 
     private Handler(Capabilities capabilities) {