code: @VisibleForTesting
      boolean runNextJob() {
          try {
              Queue<Message> messages = _messageSupplier.get();
              Message message;
  
              while ((message = messages.poll()) != null) {
                  String jobIdString = (String) message.getPayload();
  
                  // If this job has recently reported that it cannot run on this server then skip it.
                  Instant now = Instant.now();
                  Instant delayUntilTime = _recentNotOwnerDelays.get(jobIdString, EPOCH);
                  if (now.isBefore(delayUntilTime)) {
                      _log.debug("Waiting {} for next attempt to run job locally: {}",
                              Duration.between(now, delayUntilTime), jobIdString);
                      continue;
                  }
  
                  InterProcessMutex mutex = getMutex(jobIdString);
  
                  if (!acquireMutex(mutex)) {
                      _log.debug("Failed to get mutex for job {}", jobIdString);
                      continue;
                  }
                  try {
                      String jobTypeName = getJobTypeNameFromId(jobIdString);
                      RegistryEntry<?, ?> entry = _jobHandlerRegistry.getRegistryEntry(jobTypeName);
  
                      _log.info("Executing job {}... ", jobIdString);
  
                      boolean ranLocally = run(jobIdString, entry);
  
                      if (ranLocally) {
                          acknowledgeQueueMessage(message.getId());
                          _log.info("Executing job {}... DONE", jobIdString);
                      } else {
                          // The job self-reported it could not be run locally.  Cache that knowledge and wait before
                          // attempting this job again.
                          _recentNotOwnerDelays.put(jobIdString, Instant.now().plus(_notOwnerRetryDelay));
                          _recentNotOwnerDelays.cleanUp();
                          _log.info("Executing job {}... not local", jobIdString);
                      }
                  } finally {
                      mutex.release();
                  }
  
                  return true;
              }
  
              _log.debug("Job queue was empty or contained only non-local jobs");
          } catch (Throwable t) {
              _log.warn("runNextJob failed unexpectedly", t);
          }
  
          return false;
      }
docs: Dequeues the next job from the job queue and runs it.
  @return True if a job was dequeued and executed, false if the queue was empty.