apache/kafka · warning · InterruptException

Thread interrupted.

Error message

Thread interrupted.

What it means

Thrown by ShareFetchBuffer.awaitNotEmpty when the thread's interrupt status is found set at the moment the fetch timer expires. The share consumer interprets an interrupted thread as a request to abort the blocking wait for records; it surfaces the interrupt as Kafka's InterruptException so callers can handle thread cancellation cleanly rather than returning empty data.

Source

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

     * <ol>
     *     <li>The buffer was already non-empty on entry</li>
     *     <li>The buffer was populated during the wait</li>
     *     <li>The remaining time on the {@link Timer timer} elapsed</li>
     *     <li>The thread was interrupted</li>
     * </ol>
     *
     * @param timer Timer that provides time to wait
     */
    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();

View on GitHub (pinned to c31c9215e1)

Solutions

1. Catch InterruptException around poll() and treat it as a shutdown signal, then close the consumer and exit the run loop.
2. Avoid calling Thread.interrupt() on the consumer thread if you want poll() to complete; use wakeup() / close() instead to stop the share consumer cooperatively.
3. If using an ExecutorService, call shutdown() / shutdownNow() deliberately and propagate the interrupt status after cleanup.
4. Reset the interrupt flag (Thread.currentThread().interrupt()) if you must continue, then re-poll with a fresh timer.

Example fix

// before
Future<?> f = exec.submit(() -> {
    while (running) consumer.poll(Duration.ofSeconds(5));
});
f.cancel(true);  // interrupts mid-poll -> InterruptException

// after
Future<?> f = exec.submit(() -> {
    try {
        while (running) consumer.poll(Duration.ofSeconds(5));
    } catch (InterruptException | WakeupException e) {
        // graceful shutdown
    } finally {
        consumer.close();
    }
});
consumer.wakeup();  // cooperative stop instead of cancel(true)

try {
    consumer.poll(Duration.ofSeconds(30));
} catch (org.apache.kafka.common.errors.InterruptException e) {
    // The consumer thread was interrupted (e.g. shutdown). Restore the interrupt status
    // so upper layers can observe it, then exit the loop cleanly.
    Thread.currentThread().interrupt();
    log.info("Consumer poll interrupted, shutting down");
    return ConsumerPollOutcome.SHUTDOWN;
} finally {
    consumer.close();
}

When it happens

Trigger: The application thread calling consumer.poll(...) is interrupted (Thread.interrupt()) by a parent executor, shutdown hook, or CompletableFuture cancellation while the share consumer is blocked in awaitNotEmpty waiting for records to arrive.

Common situations: Shutting down an ExecutorService that owns the consumer thread; cancelling a task wrapped around poll(); Spring/@PreDestroy or container shutdown interrupting worker threads; using Future.cancel(true) on a polling task.

Related errors

• Timeout waiting for results from fetching records
• KafkaShareConsumer is not safe for multi-threaded access. cu
• Consumer is not subscribed to any topics.
• Telemetry is not enabled. Set config `${ConsumerConfig.ENABL
• Failed to close Kafka share consumer

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