apache/kafka · warning · InterruptException

Timeout waiting for results from fetching records

Error message

Timeout waiting for results from fetching records

What it means

Thrown by ShareFetchBuffer.awaitNotEmpty when the waiting thread is interrupted while blocked on the notEmpty condition. The message text says 'Timeout waiting for results' but the root cause is an InterruptedException caught during condition.await; the original interrupt is wrapped as an InterruptException to abort the poll. It indicates the blocking fetch was cancelled by an interrupt rather than by timer expiry.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetchBuffer.java:163

    void awaitNotEmpty(Timer timer) {
        lock.lock();
        try {
            while (completedFetches.isEmpty() && !wokenUp.compareAndSet(true, false)) {
                // Update the timer before we head into the loop in case it took a while to get the lock.
                timer.update();

                if (timer.isExpired()) {
                    if (Thread.interrupted())
                        throw new InterruptException("Thread interrupted.");
                    break;
                }

                if (!notEmptyCondition.await(timer.remainingMs(), TimeUnit.MILLISECONDS)) {
                    break;
                }
            }
        } catch (InterruptedException e) {
            throw new InterruptException("Timeout waiting for results from fetching records", e);
        } finally {
            lock.unlock();
            timer.update();
        }
    }

    void wakeup() {
        wokenUp.set(true);
        lock.lock();
        try {
            notEmptyCondition.signalAll();
        } finally {
            lock.unlock();
        }
    }

    /**
     * Return the set of {@link TopicIdPartition partitions} for which we have data in the buffer or pending acknowledgements.

View on GitHub (pinned to c31c9215e1)

Solutions

Use consumer.wakeup() to interrupt a poll cooperatively instead of Thread.interrupt().
If you must interrupt, handle InterruptException in the poll loop and close the consumer cleanly.
Audit executor lifecycle: prefer graceful shutdown() with a short awaitTermination before shutdownNow().
Restore the interrupt flag (Thread.currentThread().interrupt()) after handling if the thread is reused.

Example fix

// before
Thread t = new Thread(() -> consumer.poll(Duration.ofSeconds(60)));
t.start();
t.interrupt();

// after
Thread t = new Thread(() -> {
    try { consumer.poll(Duration.ofSeconds(60)); }
    catch (InterruptException e) { /* shutting down */ }
});
t.start();
consumer.wakeup();

Defensive patterns

Strategy: retry

Validate before calling

// Pre-check: confirm the consumer has not been woken up and the deadline is sane.
if (wakeupRequested.get()) {
    return ConsumerRecords.empty();
}
Duration timeout = Duration.ofMillis(Math.max(100, requestTimeoutMs));
consumer.poll(timeout);

Try / catch

try {
    records = consumer.poll(Duration.ofMillis(pollTimeoutMs));
} catch (org.apache.kafka.common.errors.InterruptException e) {
    // awaitNotEmpty was interrupted while waiting for fetch results. Decide policy:
    //  - if this is a graceful shutdown signal: re-interrupt and exit;
    //  - otherwise: brief back-off and retry the poll.
    if (shutdownRequested.get()) {
        Thread.currentThread().interrupt();
        return ConsumerRecords.empty();
    }
    Thread.sleep(backoffMs);
}

Prevention

Distinguish 'interrupted for shutdown' from 'interrupted spuriously' using your own shutdown flag.
Use consumer.wakeup() to cancel a poll from another thread instead of interrupting the thread.
Avoid blocking the consumer thread in user code between poll() calls.

When it happens

Trigger: Thread.interrupt() is invoked on the consumer thread while it is blocked in notEmptyCondition.await(...) inside awaitNotEmpty; the catch(InterruptedException) branch wraps the cause into InterruptException and rethrows.

Common situations: Same family as error 223: executor shutdown, Future.cancel(true), container-driven thread interruption, or application code calling interrupt() on the polling thread while it is waiting for records to land in the buffer.

Related errors

AI-assisted analysis of apache/kafka@c31c9215e1 (2026-08-03). Data as JSON: /data/errors/fb32c7e63ec7f47f.json. Report an issue: GitHub.