*
   * <p>Note: If {@code iterable} is a {@link Queue}, the returned iterable will instead use {@link
   * Queue#isEmpty} and {@link Queue#remove()}, since {@link Queue}'s iteration order is undefined.
   * Calling {@link Iterator#hasNext()} on a generated iterator from the returned iterable may cause
   * an item to be immediately dequeued for return on a subsequent call to {@link Iterator#next()}.
   *

    // Cause acceptConnections() to break out.
    serverSocket.closeQuietly()

    // Await shutdown.
    for (queue in taskRunner.activeQueues()) {
      if (!queue.idleLatch().await(5, TimeUnit.SECONDS)) {
        throw AssertionError("Gave up waiting for queue to shut down")
      }
    }
    taskRunnerBackend.shutdown()
  }

  private fun serveConnection(
    connectionIndex: Int,

- `process_start_time_seconds` has been added to kubelet’s '/metrics/probes' endpoint ([#77975](https://github.com/kubernetes/kubernetes/pull/77975), [@logicalhan](https://github.com/logicalhan))

    startNs = queue.taskRunner.backend.nanoTime()
    log(task, queue, "starting")
  }

  var completedNormally = false
  try {
    val result = block()
    completedNormally = true
    return result
  } finally {
    if (loggingEnabled) {
      val elapsedNs = queue.taskRunner.backend.nanoTime() - startNs
      if (completedNormally) {
        log(task, queue, "finished run in ${formatDuration(elapsedNs)}")

  }

  @Test fun taskFailsWithUncheckedException() {
    queue.schedule("task", TimeUnit.MILLISECONDS.toNanos(100)) {
      log.put("failing task running")
      throw RuntimeException("boom!")
    }

    queue.schedule("task", TimeUnit.MILLISECONDS.toNanos(200)) {
      log.put("normal task running")
      return@schedule -1L
    }

    queue.idleLatch().await(500, TimeUnit.MILLISECONDS)

 * }
 * </pre>
 */
public class ESSourceReader implements DocumentReader {
    private static final Logger logger = LogManager.getLogger(ESSourceReader.class);

    /** Queue of documents. */
    protected final Queue<Map<String, Object>> queue = new ConcurrentLinkedQueue<>();
    /** Flag indicating if reading is finished. */
    protected final AtomicBoolean isFinished = new AtomicBoolean(false);
    /** Random number generator. */

        this.filterIndex = filterIndex;
    }

    /**
     * Returns the number of queue shards.
     * @return The number of queue shards.
     */
    public int getQueueShards() {
        return queueShards;
    }

    /**
     * Sets the number of queue shards.
     * @param queueShards The number of queue shards.
     */
    public void setQueueShards(final int queueShards) {

/**
 * The UrlQueue interface represents a queue of URLs to be processed by a web crawler.
 * It provides methods to get and set various properties of a URL queue entry.
 *
 * @param <IDTYPE> the type of the identifier for the URL queue entry
 */
public interface UrlQueue<IDTYPE> {

    /**
     * Retrieves the unique identifier of the URL queue.
     *
     * @return the unique identifier of type IDTYPE.
     */

    private final BlockingQueue<String> queue;

    EnableWrites(BlockingQueue<String> queue, long tMinus) {
      super(tMinus);
      assertFalse(queue.isEmpty());
      assertFalse(queue.offer("shouldBeRejected"));
      this.queue = queue;
    }

    @Override
    protected void doAction() {
      assertNotNull(queue.remove());
    }
  }

  @Param({"100", "10000"})
  private int size;

  @Param private HeapType heap;

  private Queue<Integer> queue;

  private final Random random = new Random();

  @BeforeExperiment
  void setUp() {
    queue = heap.create(comparator.get());
    for (int i = 0; i < size; i++) {
      queue.add(random.nextInt());
    }
  }

  @Benchmark
  void pollAndAdd(int reps) {