apache/kafka · critical · java.util.ConcurrentModificationException

KafkaConsumer is not safe for multi-threaded access. current

Error message

KafkaConsumer is not safe for multi-threaded access. currentThread(name: {}, id: {}) otherThread(id: {})

What it means

Thrown by acquire() when the consumer's light-weight single-writer lock detects that the calling thread differs from the thread that currently holds the consumer (and the slot is not free). KafkaConsumer is intentionally not thread-safe; it tracks an owning thread id via an AtomicLong and rejects concurrent access with a ConcurrentModificationException rather than blocking. The message names both threads to make the misuse debuggable.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:2229

            metadata.maybeThrowBootstrapFatalException();
        } catch (RuntimeException e) {
            release();
            throw e;
        }
    }

    /**
     * Acquire the light lock protecting this consumer from multithreaded access. Instead of blocking
     * when the lock is not available, however, we just throw an exception (since multithreaded usage is not
     * supported).
     *
     * @throws ConcurrentModificationException if another thread already has the lock
     */
    private void acquire() {
        final Thread thread = Thread.currentThread();
        final long threadId = thread.getId();
        if (threadId != currentThread.get() && !currentThread.compareAndSet(NO_CURRENT_THREAD, threadId))
            throw new ConcurrentModificationException("KafkaConsumer is not safe for multi-threaded access. " +
                "currentThread(name: " + thread.getName() + ", id: " + threadId + ")" +
                " otherThread(id: " + currentThread.get() + ")"
            );
        refCount.incrementAndGet();
    }

    /**
     * Release the light lock protecting the consumer from multithreaded access.
     */
    private void release() {
        if (refCount.decrementAndGet() == 0)
            currentThread.set(NO_CURRENT_THREAD);
    }

    private void subscribeInternal(Pattern pattern, Optional<ConsumerRebalanceListener> listener) {
        acquireAndEnsureOpen();
        try {
            throwIfGroupIdNotDefined();

View on GitHub (pinned to c31c9215e1)

Solutions

Confine all consumer method calls to a single dedicated thread; have other threads submit work via a queue and call consumer.wakeup() (the only thread-safe method) to unblock poll.
If you need parallel consumption, create one KafkaConsumer instance per thread, each joining the same group.
Move commit calls into the poll-loop thread (use auto-commit or commitAsync inside the loop) rather than a separate thread.
Wrap the consumer in a thread-affine accessor that asserts Thread.currentThread() == ownerThread before delegating.

Example fix

// before
KafkaConsumer<String,String> c = new KafkaConsumer<>(props);
ExecutorService es = Executors.newFixedThreadPool(2);
es.submit(() -> c.poll(Duration.ofMillis(100)));
es.submit(() -> c.commitSync()); // throws ConcurrentModificationException

// after
KafkaConsumer<String,String> c = new KafkaConsumer<>(props);
Thread consumerThread = new Thread(() -> {
    while (!closed) {
        ConsumerRecords<String,String> recs = c.poll(Duration.ofMillis(100));
        // process, then commit on THIS thread
        if (!recs.isEmpty()) c.commitSync();
    }
});
consumerThread.start();
// other threads only call c.wakeup() to signal shutdown

Defensive patterns

Strategy: validation

Validate before calling

// Enforce single-thread ownership at the application level:
private final Thread ownerThread = Thread.currentThread();
private void ensureOwner() {
    if (Thread.currentThread() != ownerThread)
        throw new IllegalStateException("Consumer accessed from non-owner thread");
}
// Call ensureOwner() before every consumer method.
// For cross-thread wake-up, use consumer.wakeup() only — it is thread-safe.

Type guard

// Java: bind consumer to a single thread
static boolean isOwnerThread(Thread owner) {
    return Thread.currentThread() == owner;
}

Try / catch

try {
    consumer.poll(Duration.ofMillis(100));
} catch (ConcurrentModificationException e) {
    log.error("Consumer used from multiple threads", e);
    // Do NOT retry on the same thread — fix the design instead.
    throw e;
}

Prevention

KafkaConsumer is NOT thread-safe. Create it on, and use it only from, one thread.
For multi-thread consumption, run one KafkaConsumer per thread (or use Kafka Streams).
To interrupt a poll from another thread, call consumer.wakeup() — that is the only thread-safe method.
Wrap the consumer in a thread-affine facade whose methods assert Thread.currentThread() == ownerThread.
Never put a KafkaConsumer in a shared singleton, Spring @Component with singleton scope, or static field accessed by multiple threads.

When it happens

Trigger: Two application threads both calling methods on the same KafkaConsumer instance (e.g. a poll thread and a commit thread); waking the consumer via wakeup() is allowed, but any other method called from a non-owner thread throws; shared consumer passed to multiple executors/schedulers.

Common situations: Wrapping the consumer in a shared service bean injected into multiple controllers; scheduling commitAsync from a different ScheduledExecutorService than the poll loop; Spring Kafka ListenerContainer sharing a consumer with application code; migrations from a queue library that permitted concurrent consumers.

Related errors

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