code: private CompletableFuture<ContainerHandle> startContainerInternal(int containerId) {
          ContainerWithHandle newContainer = new ContainerWithHandle(this.factory.createStreamSegmentContainer(containerId),
                  new SegmentContainerHandle(containerId));
          ContainerWithHandle existingContainer = this.containers.putIfAbsent(containerId, newContainer);
          if (existingContainer != null) {
              // We had multiple concurrent calls to start this Container and some other request beat us to it.
              newContainer.container.close();
              throw new IllegalArgumentException(String.format("Container %d is already registered.", containerId));
          }
  
          log.info("Registered SegmentContainer {}.", containerId);
  
          // Attempt to Start the container, but first, attach a shutdown listener so we know to unregister it when it's stopped.
          Services.onStop(
                  newContainer.container,
                  () -> unregisterContainer(newContainer),
                  ex -> handleContainerFailure(newContainer, ex),
                  this.executor);
          return Services.startAsync(newContainer.container, this.executor)
                         .thenApply(v -> newContainer.handle);
      }
docs: Creates a new Container and attempts to register it. This method works in an optimistic manner: it creates the
  Container first and then attempts to register it, which should prevent us from having to lock on this entire method.
  Creating new containers is cheap (we don't start them yet), so this operation should not take any extra resources.
  
  @param containerId The Id of the Container to start.
  @return A CompletableFuture which will be completed with a ContainerHandle once the container has been started.