code: @Override
    public Future<? extends List<Pair<SpecExecutor.Verb, Spec>>> changedSpecs() {
      List<Pair<SpecExecutor.Verb, Spec>> changesSpecs = new ArrayList<>();
  
      try {
        Pair<SpecExecutor.Verb, Spec> specPair = _jobSpecQueue.take();
        _metrics.jobSpecDeqCount.incrementAndGet();
        do {
          changesSpecs.add(specPair);
  
          // if there are more elements then pass them along in this call
          specPair = _jobSpecQueue.poll();
        } while (specPair != null);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
  
      return new CompletedFuture(changesSpecs, null);
    }
docs: This method returns job specs receive from Kafka. It will block if there are no job specs.
  @return list of (verb, jobspecs) pairs.