code: @Synchronized
      private void checkAssignment() {
          long traceId = LoggerHelpers.traceEnter(log, "checkAssignment");
          try {
              Exceptions.checkNotClosed(closed.get(), this);
  
              // Fetch the list of containers that is supposed to be owned by this host.
              Set<Integer> desiredList = getDesiredContainerList();
              if (desiredList != null) {
                  Collection<Integer> runningContainers = new HashSet<>(this.handles.keySet());
                  Collection<Integer> containersPendingTasks = new HashSet<>(this.pendingTasks);
  
                  // Filter out containers which have pending tasks so we don't initiate conflicting events on the same
                  // containers. Events for these containers will be tried on subsequent runs of this executor.
                  Collection<Integer> containersToBeStarted = CollectionHelpers.filterOut(desiredList, runningContainers);
                  containersToBeStarted = CollectionHelpers.filterOut(containersToBeStarted, containersPendingTasks);
  
                  Collection<Integer> containersToBeStopped = CollectionHelpers.filterOut(runningContainers, desiredList);
                  containersToBeStopped = CollectionHelpers.filterOut(containersToBeStopped, containersPendingTasks);
  
                  log.info("Container Changes: Desired = {}, Current = {}, PendingTasks = {}, ToStart = {}, ToStop = {}.",
                          desiredList, runningContainers, containersPendingTasks, containersToBeStarted,
                          containersToBeStopped);
  
                  // Initiate the start and stop tasks asynchronously.
                  containersToBeStarted.forEach(this::startContainer);
                  containersToBeStopped.forEach(this::stopContainer);
              } else {
                  log.warn("No segment container assignments found");
              }
          } catch (Throwable e) {
              // Need to catch all exceptions here since throwing any exception here will halt this scheduled job.
              log.warn("Failed to monitor the segmentcontainer assignment: ", e);
          } finally {
              LoggerHelpers.traceLeave(log, "checkAssignment", traceId);
          }
      }
docs: The container assignment monitor.
  This method will fetch the current owned containers for this host and ensures that the local containers' state
  reflects this.