apache/kafka · error · InterruptException

Interrupted waiting for results for application event ${even

Error message

Interrupted waiting for results for application event ${event}

What it means

Thrown as InterruptException (a KafkaException subtype) by ApplicationEventHandler.addAndGet when Thread.interrupted() returns true after the event has been added to the network-thread queue but before getResult blocks. It is a pre-wait interrupt check: the client surfaces a pending interrupt immediately instead of letting it surface later inside the blocking future.get(). This is the async-consumer application-thread → network-thread bridge.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java:145

     * Add a {@link CompletableApplicationEvent} to the handler. The method blocks waiting for the result, and will
     * return the result value upon successful completion; otherwise throws an error.
     *
     * <p/>
     *
     * See {@link ConsumerUtils#getResult(Future)} for more details.
     *
     * @param event A {@link CompletableApplicationEvent} created by the polling thread
     * @return      Value that is the result of the event
     * @param <T>   Type of return value of the event
     */
    public <T> T addAndGet(final CompletableApplicationEvent<T> event) {
        Objects.requireNonNull(event, "CompletableApplicationEvent provided to addAndGet must be non-null");
        add(event);
        // Check if the thread was interrupted before we start waiting, to ensure that we
        // propagate the exception even if we end up not having to wait (the event could complete
        // between the time it's added and the time we attempt to getResult)
        if (Thread.interrupted()) {
            throw new InterruptException("Interrupted waiting for results for application event " + event);
        }
        return ConsumerUtils.getResult(event.future());
    }

    @Override
    public void close() {
        close(Duration.ZERO);
    }

    public void close(final Duration timeout) {
        closer.close(
                () -> Utils.closeQuietly(() -> networkThread.close(timeout), "consumer network thread"),
                () -> log.warn("The application event handler was already closed")
        );
    }

    /**
     * Best-effort check that the consumer network thread is still alive. If the thread has

View on GitHub (pinned to c31c9215e1)

Solutions

Stop interrupting the consumer thread — prefer consumer.close(duration) and the consumer's own poll/request timeouts for cancellation.
If interrupts are expected (shutdown), catch InterruptException, restore the interrupt flag (Thread.currentThread().interrupt()), and exit cleanly.
Avoid sharing a consumer across threads; own it on a single thread so interrupts come from your own lifecycle code.

Example fix

// before
executor.shutdownNow(); // sends Thread.interrupt() to the polling thread

// after
consumer.wakeup();
executor.shutdown();
awaitTermination(executor, Duration.ofSeconds(30));
// inside the consumer loop:
try {
    ConsumerRecords<?,?> recs = consumer.poll(Duration.ofMillis(500));
} catch (InterruptException | WakeupException e) {
    Thread.currentThread().interrupt();
    return; // graceful shutdown
}

Defensive patterns

Strategy: try-catch

Validate before calling

// Avoid interrupting the consumer thread from outside; signal shutdown via a flag and close().
private volatile boolean running = true;
// in the consumer loop:
while (running) { consumer.poll(Duration.ofMillis(500)); }
// shutdown path: running=false; consumer.wakeup();  -- do NOT Thread.interrupt() the consumer thread

Type guard

import org.apache.kafka.common.errors.InterruptException;

/** True iff a throwable is Kafka's InterruptException wrapper around thread interruption. */
static boolean isKafkaInterrupt(Throwable t) {
    return t instanceof InterruptException;
}

Try / catch

try {
    consumer.poll(Duration.ofMillis(1000));
} catch (org.apache.kafka.common.errors.InterruptException e) {
    // Honor the interrupt: restore the flag so callers up the stack see it.
    Thread.currentThread().interrupt();
    // then exit the loop cleanly — do not swallow
    running = false;
}

Prevention

Never call Thread.interrupt() on the consumer thread; use consumer.wakeup() for cooperative shutdown.
In shutdown hooks, set a stop flag and wakeup() — don't interrupt from a signal handler.
If you must catch InterruptException, always re-set the interrupt status via Thread.currentThread().interrupt().
Close the consumer (KafkaConsumer.close(timeout)) in a finally block to release the network thread cleanly.

When it happens

Trigger: The application thread calling the consumer is interrupted (Thread.interrupt()) while invoking any blocking consumer API that goes through addAndGet (e.g. committed, partitionsFor, listTopics, position, offsets-for-times, beginning/end offsets in the new async consumer). The interrupt flag was set before the consumer had a chance to block on the future.

Common situations: Application frameworks that interrupt worker threads on shutdown (Spring @PreDestroy, executor.shutdownNow(), servlet container unload); timeouts implemented via Thread.interrupt() rather than the consumer's own API timeouts; a parent thread canceling a task that wraps a consumer call; mixed use of the consumer inside a ForkJoinPool whose tasks get cancelled.

Related errors

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