> ## Content Index > Fetch the complete content index at: https://www.ggorantala.dev/llms.txt > Use this file to discover other available public pages before exploring further. # Fix KafkaConsumer Is Not Safe for Multi-Threaded Access - URL: https://www.ggorantala.dev/kafka-consumer-not-safe-for-multi-threaded-access/ - Published: 2026-03-06T14:36:00.000Z - Updated: 2026-08-29T08:20:57.000Z - Description: Hitting ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access? Here's why the thread guard fires and the correct fix. - Author: Gopi Gorantala - Tags: kafka-errors, concurrentmodificationexception, kafka-consumer, Java, kafka-threading, spring-kafka, event-streaming ## 1\. The Error ```text Exception in thread "pool-1-thread-3" java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access. currentThread(name: pool-1-thread-3, id: 27) otherThread(id: 1) at org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.acquire(ClassicKafkaConsumer.java:1832) at org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.acquireAndEnsureOpen(ClassicKafkaConsumer.java:1816) at org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.commitSync(ClassicKafkaConsumer.java:707) at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1149) at com.example.OrderProcessor.lambda$process$0(OrderProcessor.java:52) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ``` On kafka-clients 2.7 and earlier the frame is `KafkaConsumer.acquire` directly (the `internals.ClassicKafkaConsumer` split landed in 3.7 with KIP-848 groundwork). Before 2.8 the message also lacks the `currentThread(...) otherThread(...)` suffix, which makes older versions much harder to debug — the thread IDs were added precisely because everyone hits this. Despite the class name, this has nothing to do with `java.util` collections being modified during iteration. Kafka reuses `ConcurrentModificationException` for its own thread-ownership guard. **Typical setup:** kafka-clients 3.x (any transport — Docker local, k8s, MSK, Confluent Cloud; the broker is irrelevant, this is 100% client-side), raw `KafkaConsumer` with a thread pool for record processing, or a shutdown hook calling `consumer.close()`. The same guard fires from other entry points depending on which method you called from the wrong thread: `poll()`, `commitAsync()`, `seek()`, `position()`, `pause()`, `resume()`, `close()` — every public method except `wakeup()`. ## 2\. How to Reproduce It (step-by-step) Single-broker KRaft cluster: ```yaml # docker-compose.yml services: kafka: image: apache/kafka:3.8.0 ports: - "9092:9092" environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:29093 KAFKA_LISTENERS: PLAINTEXT://:29092,CONTROLLER://:29093,EXTERNAL://:9092 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 ``` ```bash docker compose up -d docker compose exec kafka /opt/kafka/bin/kafka-topics.sh \ --bootstrap-server localhost:29092 --create --topic orders --partitions 3 docker compose exec kafka /opt/kafka/bin/kafka-producer-perf-test.sh \ --topic orders --num-records 1000 --record-size 200 --throughput -1 \ --producer-props bootstrap.servers=localhost:29092 ``` The classic mistake — poll on the main thread, process **and commit** on a worker pool: ```java // kafka-clients 3.8.0 Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "order-processor"); props.put("enable.auto.commit", "false"); props.put("key.deserializer", StringDeserializer.class.getName()); props.put("value.deserializer", StringDeserializer.class.getName()); KafkaConsumer consumer = new KafkaConsumer<>(props); ExecutorService pool = Executors.newFixedThreadPool(4); consumer.subscribe(List.of("orders")); while (true) { ConsumerRecords records = consumer.poll(Duration.ofMillis(500)); for (ConsumerRecord rec : records) { pool.submit(() -> { process(rec); // BOOM: commit from a worker thread consumer.commitSync(Map.of( new TopicPartition(rec.topic(), rec.partition()), new OffsetAndMetadata(rec.offset() + 1))); }); } } ``` Run it against the seeded topic and the exception fires within the first batch — no load, no timing luck required. If it seems intermittent in your code, it's because the guard is a check against *concurrent* entry: it only throws while the poll thread is simultaneously inside a consumer method. A worker-thread commit that lands while the poll thread is parked between `poll()` calls can slip through silently — which is worse, because now you have an unsynchronized data race that "works" in dev. **Other common triggers:** - A JVM shutdown hook or `SIGTERM` handler calling `consumer.close()` while the poll loop is blocked in `poll()`. - A metrics/health endpoint (Micrometer gauge, actuator) calling `consumer.metrics()` or `consumer.position()` from an HTTP thread. - Spring Kafka: stashing the `Consumer` parameter from a `@KafkaListener` method into a field and touching it from a scheduler or REST controller. - A rebalance-listener or retry framework invoking `seek()` from an async callback. ## 3\. Why It Happens — Surface Level `KafkaConsumer` is deliberately **not** thread-safe, and instead of letting you corrupt its state it fail-fasts. Every public method first calls an internal `acquire()`, which records the calling thread's ID. If a second thread enters while another thread is still inside the consumer, `acquire()` throws `ConcurrentModificationException` immediately. Exactly one method is exempt and documented thread-safe: `wakeup()`. Everything else — `poll`, `commitSync`, `commitAsync`, `seek`, `pause`, `close`, even `metrics()` — must be called from the single thread that owns the consumer. ## 4\. Why It Happens — Under the Hood The guard is not a lock. It's a CAS-based ownership check (unchanged in spirit from 0.9 through the 3.9/4.x `ClassicKafkaConsumer`): ```java // org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer (abridged) private static final long NO_CURRENT_THREAD = -1L; private final AtomicLong currentThread = new AtomicLong(NO_CURRENT_THREAD); private final AtomicInteger refCount = new AtomicInteger(0); 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()` decrements `refCount` and resets `currentThread` to `NO_CURRENT_THREAD` at zero. Two consequences worth internalizing: **It detects concurrency, not foreign ownership.** The consumer doesn't remember "thread 1 owns me." It only rejects a thread entering *while another thread is inside*. Alternating single-threaded access from different threads passes the guard — and is still broken, because the consumer's internal state (`SubscriptionState`, the fetch buffer, the coordinator's request queue) is plain unsynchronized mutable state with no memory-visibility guarantees across threads. The exception is the *friendly* failure mode. The unfriendly ones are silently lost seeks, committed offsets for records you never processed, and `IllegalStateException: No current assignment for partition` when one thread's view of the assignment is stale. **Why the design is single-threaded at all.** Everything in the classic consumer rides on the poll thread: the fetcher's network I/O, offset commit responses, and — critically — the `ConsumerRebalanceListener` callbacks and `JoinGroup`/`SyncGroup` handling all execute inside `poll()`. Since KIP-62 (0.10.1) there *is* a background heartbeat thread, but it talks to the coordinator through internal synchronized structures, never through the public API — its existence is why people wrongly assume the consumer must be thread-safe. The new `AsyncKafkaConsumer` (`group.protocol=consumer`, KIP-848, GA in 4.0) moves networking to a background event loop, but the public API keeps the exact same single-threaded contract and the same `acquire()` guard. The threading rule is not a legacy wart that KIP-848 fixes; it's the contract. ## 5\. The Fix ### Fix 1: Commit on the poll thread (correct for most workloads) Workers process; the poll thread owns all consumer interaction. Hand completed offsets back through a thread-safe structure: ```diff KafkaConsumer consumer = new KafkaConsumer<>(props); ExecutorService pool = Executors.newFixedThreadPool(4); +Map pending = new ConcurrentHashMap<>(); consumer.subscribe(List.of("orders")); while (true) { ConsumerRecords records = consumer.poll(Duration.ofMillis(500)); for (ConsumerRecord rec : records) { pool.submit(() -> { process(rec); - consumer.commitSync(Map.of( - new TopicPartition(rec.topic(), rec.partition()), - new OffsetAndMetadata(rec.offset() + 1))); + pending.merge( + new TopicPartition(rec.topic(), rec.partition()), + new OffsetAndMetadata(rec.offset() + 1), + (a, b) -> b.offset() > a.offset() ? b : a); }); } + if (!pending.isEmpty()) { + Map snapshot = new HashMap<>(pending); + snapshot.forEach((tp, om) -> pending.remove(tp, om)); + consumer.commitAsync(snapshot, (offsets, ex) -> { + if (ex != null) log.warn("Commit failed, will retry on next cycle", ex); + }); + } } ``` Caveat you must own consciously: committing the max completed offset per partition means an out-of-order worker failure can mark earlier records as done (at-most-once for those records). If you need strict at-least-once with parallel workers, track per-partition contiguous completion (commit only up to the highest offset with no gaps below it) — or use a framework that does (see §6). ### Fix 2: Shutdown — `wakeup()` is the only cross-thread call you get ```diff -Runtime.getRuntime().addShutdownHook(new Thread(consumer::close)); +final AtomicBoolean running = new AtomicBoolean(true); +Runtime.getRuntime().addShutdownHook(new Thread(() -> { + running.set(false); + consumer.wakeup(); // thread-safe by contract; poll() throws WakeupException +})); try { - while (true) { + while (running.get()) { ConsumerRecords records = consumer.poll(Duration.ofMillis(500)); ... } +} catch (WakeupException e) { + // expected on shutdown } finally { consumer.close(Duration.ofSeconds(10)); // poll thread: leaves group cleanly } ``` Closing on the poll thread matters beyond avoiding the exception: `close()` sends `LeaveGroup`, so the partition handoff happens immediately instead of after `session.timeout.ms`. ### Fix 3: Spring Kafka — never let the `Consumer` escape the listener thread ```diff @KafkaListener(topics = "orders") -public void listen(ConsumerRecord rec, Consumer consumer) { - this.stashedConsumer = consumer; // touched later by a @Scheduled method → guard fires +public void listen(ConsumerRecord rec, Acknowledgment ack) { process(rec); + ack.acknowledge(); // executed on the container's consumer thread } ``` For out-of-band operations (pause on backpressure, seek from an admin endpoint) use the container, not the consumer: `registry.getListenerContainer("id").pause()` is thread-safe and applied by the container's polling thread on its next cycle. ## 6\. Best Practices & The Better Design The design that avoids this class of bug entirely: **one thread owns each consumer, full stop.** Parallelism comes from either more consumers or a decoupled worker stage — never from sharing. **Option A — consumer-per-thread.** Simplest correct model. N threads, each with its own `KafkaConsumer`, same `group.id`. Scaling ceiling = partition count. This is exactly what Spring Kafka's `ConcurrentMessageListenerContainer` gives you with `concurrency=N` — each of the N containers has a dedicated consumer and thread, which is why the threading problem simply never surfaces in idiomatic Spring Kafka code. **Option B — poll thread + worker pool with pause/resume backpressure.** When per-record work is slow (calls to downstream services) and you'd need more parallelism than partitions: ```java // The poll thread is the ONLY thread touching `consumer`. while (running.get()) { ConsumerRecords records = consumer.poll(Duration.ofMillis(200)); if (!records.isEmpty()) { inFlight.addAll(submitToWorkers(records)); // workers never see the consumer consumer.pause(consumer.assignment()); // stop fetching, keep polling } drainCompleted(inFlight, pending); // collect finished offsets if (inFlight.isEmpty()) consumer.resume(consumer.paused()); commitContiguous(pending, consumer); // commitAsync on poll thread } ``` The `pause()`/`resume()` trick is the load-bearing part: the loop keeps calling `poll()` so heartbeats, rebalance callbacks, and (pre-KIP-62-style) liveness via `max.poll.interval.ms` stay healthy while workers grind — without fetching records you can't handle yet. This is the pattern that also kills the rebalance-timeout failure mode, because poll-loop latency no longer depends on processing latency. **Option C — use something that implements Option B for you.** Confluent's `parallel-consumer` (per-key ordering + contiguous offset tracking), Spring's `@KafkaListener` with `AckMode.MANUAL`, or Kafka Streams if the work is transformation-shaped. In 2026 there is little reason to hand-roll the offset-gap bookkeeping for a business service. ## 7\. How to Prevent It Long-Term - **Convention: the consumer variable never leaves the method that created it.** No fields, no getters, no passing into callbacks. Make this a code-review rule; it's greppable (`private.*KafkaConsumer` as a field is a smell outside a runnable's scope). - **Static analysis.** Error Prone / ArchUnit rule forbidding `KafkaConsumer` references from `@Scheduled`, `@RestController`, or `Runnable` classes catches the stash-and-touch pattern in CI. - **Test under real concurrency.** A Testcontainers integration test that runs your consumer loop while hammering your health/metrics endpoints and then sends `SIGTERM` will surface both the guard exception and the shutdown-hook variant before production does. - **Don't disable the pain signal.** The race that slips past the guard (alternating threads) shows up as duplicate or skipped processing. Alert on committed-offset regressions and on `records-consumed-rate` vs downstream throughput divergence — those are the symptoms of the silent version of this bug. - **Prefer framework-managed containers.** If every service hand-rolls its poll loop, every service re-solves shutdown, backpressure, and commit ordering. Standardize on Spring Kafka containers or parallel-consumer across the team and this exception disappears from your logs org-wide. ## 8\. Key Takeaways - `ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access` is Kafka's fail-fast thread-ownership guard (`acquire()` CAS check), not a collections bug — every public consumer method except `wakeup()` must run on the one thread that owns the consumer. - The guard only fires on truly *concurrent* entry. Alternating access from multiple threads passes the check and silently corrupts state — the exception is the good outcome. - Most common triggers: committing from a worker pool, `consumer.close()` in a shutdown hook, metrics endpoints calling `consumer.metrics()`/`position()`, and stashed `Consumer` references in Spring listeners. - Fix = route every consumer call back to the poll thread: offset handoff via a concurrent map, shutdown via `wakeup()` \+ flag, Spring via `Acknowledgment` and container-level pause/resume. - The single-threaded contract is unchanged in the KIP-848 `AsyncKafkaConsumer` — design for one-thread-per-consumer (or use parallel-consumer / Spring containers) rather than waiting for a thread-safe client that isn't coming.