apache/kafka · error · 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 as a ConcurrentModificationException from ClassicKafkaConsumer.acquire() (ClassicKafkaConsumer.java:1253) when a consumer method is entered from a thread that does not own the consumer's internal lock. KafkaConsumer is intentionally not thread-safe; instead of blocking on a second caller, the per-instance AtomicLong holding the owning thread id is compare-and-set, and on failure the library throws immediately to protect mutable state (subscriptions, fetch buffers, coordinator state, refcount). The only thread-safe method is wakeup(); every other public method (poll, subscribe, commit, seek, close, etc.) routes through acquire()/acquireAndEnsureOpen().

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java:1257

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

    /**
     * Acquire the light lock protecting this consumer from multi-threaded access. Instead of blocking
     * when the lock is not available, however, we just throw an exception (since multi-threaded 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 multi-threaded access.
     */
    private void release() {
        if (refcount.decrementAndGet() == 0)
            currentThread.set(NO_CURRENT_THREAD);
    }

    private void throwIfNoAssignorsConfigured() {
        if (assignors.isEmpty())
            throw new IllegalStateException("Must configure at least one partition assigner class name to " +
                    ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG + " configuration property");

View on GitHub (pinned to c31c9215e1)

Solutions

Restrict all consumer calls to the single thread that created the consumer; have other threads enqueue work items and let the owning thread execute them.
Never touch the consumer inside commitAsync's callback — only record the result (a future/atomic) and act on the owning thread.
To interrupt a blocked poll() from another thread, use consumer.wakeup() (the one thread-safe method); the owner thread then sees WakeupException and can shut down cleanly.
If concurrent processing is genuinely required, run one KafkaConsumer instance per thread and partition work externally rather than sharing one instance.

Example fix

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

// after — only the owning thread touches the consumer
while (running) {
    ConsumerRecords<String,String> recs = c.poll(Duration.ofSeconds(1));
    recs.forEach(r -> es.submit(() -> handle(r)));   // processing off-thread is fine
    c.commitSync();                                   // still on the owning thread
}

Defensive patterns

Strategy: validation

Validate before calling

// Pin the consumer to its creating thread; validate before every call
Thread owner = consumerOwnerThread; // captured at `new KafkaConsumer<>(...)`
if (Thread.currentThread() != owner) {
    throw new IllegalStateException(
        "KafkaConsumer touched from " + Thread.currentThread().getName() +
        ", expected " + owner.getName());
}
consumer.poll(Duration.ofSeconds(1));

Type guard

// Thread-affine wrapper: every delegated method asserts single-thread ownership
static <K, V> Consumer<K, V> confineToCurrentThread(Consumer<K, V> c) {
    final Thread owner = Thread.currentThread();
    return new Consumer<K, V>() {
        private void check() {
            if (Thread.currentThread() != owner)
                throw new IllegalStateException("consumer bound to " + owner);
        }
        public ConsumerRecords<K, V> poll(Duration t) { check(); return c.poll(t); }
        // ... delegate every other method through check() ...
        public void wakeup() { c.wakeup(); } // the ONLY thread-safe method
    };
}

Try / catch

try {
    consumer.poll(Duration.ofSeconds(1));
} catch (ConcurrentModificationException e) {
    // Do NOT retry from this thread; hand the work to the owning thread
    // or replace with one-consumer-per-thread design.
    throw new IllegalStateException("consumer misused across threads", e);
}

Prevention

Create the KafkaConsumer on the exact thread that will call poll/commit/seek on it.
Never share a consumer instance across threads; give each worker thread its own consumer.
For cross-thread signaling use consumer.wakeup() — it is the only thread-safe method.
Hand records off to other threads via a queue; do not let them touch the consumer.

When it happens

Trigger: A thread other than the consumer's owner thread invokes any consumer method while the owner thread is still inside one (e.g. poll still running on thread A, thread B calls commitSync or position). All public methods are guarded by acquire() at ClassicKafkaConsumer.java:1123 (close) and the acquireAndEnsureOpen() helper used by poll/subscribe/seek/commit.

Common situations: Sharing a single consumer across an ExecutorService or thread pool; calling consumer methods from inside a commitAsync callback (which runs on a different thread); a manual 'heartbeat' or 'offset flusher' thread; a ConsumerRebalanceListener that calls consumer methods from the rebalance thread; close() racing with poll(); framework code (Spring, Quarkus) that hands the consumer to worker threads.

Related errors

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