boolean runOnce() {
        // first blocking call = get next task. blocks only if there are no active splits and queued
        // tasks.
        SplitFetcherTask task;
        lock.lock();
        try {
            if (closed) {
                return false;
            }

            task = getNextTaskUnsafe();
            if (task == null) {
                // (spurious) wakeup, so just repeat
                return true;
            }

            LOG.debug("Prepare to run {}", task);
            // store task for #wakeUp
            this.runningTask = task;
        } finally {
            lock.unlock();
        }

        // execute the task outside of lock, so that it can be woken up
        boolean taskFinished;
        try {
            taskFinished = task.run();
        } catch (Exception e) {
            throw new RuntimeException(
                    String.format(
                            "SplitFetcher thread %d received unexpected exception while polling the records",
                            id),
                    e);
        }

        // re-acquire lock as all post-processing steps, need it
        lock.lock();
        try {
            this.runningTask = null;
            processTaskResultUnsafe(task, taskFinished);
        } finally {
            lock.unlock();
        }
        return true;
    }
@Test
    public void testNotifiesWhenGoingIdleConcurrent() throws Exception {
        final FutureCompletingBlockingQueue<RecordsWithSplitIds<Object>> queue =
                new FutureCompletingBlockingQueue<>();
        final SplitFetcher<Object, TestingSourceSplit> fetcher =
                createFetcherWithSplit(
                        "test-split",
                        queue,
                        new TestingSplitReader<>(finishedSplitFetch("test-split")));

        final QueueDrainerThread queueDrainer = new QueueDrainerThread(queue, fetcher, 1);
        queueDrainer.start();

        fetcher.runOnce();

        queueDrainer.sync();

        // either we got the notification that the fetcher went idle after the queue was drained
        // (thread finished)
        // or the fetcher was already idle when the thread drained the queue (then we need no
        // additional notification)
        assertThat(queue.getAvailabilityFuture().isDone() || queueDrainer.wasIdleWhenFinished())
                .isTrue();
    }