Antonio Gagliardi created KAFKA-21435:
-----------------------------------------
Summary: MirrorCheckpointTask.stop() closes its stores while the
scheduler is still starting them
Key: KAFKA-21435
URL: https://issues.apache.org/jira/browse/KAFKA-21435
Project: Kafka
Issue Type: Bug
Components: mirrormaker
Affects Versions: 4.4.0, 4.3.0, 4.2.0, 4.1.0, 4.0.0, 3.9.0
Reporter: Antonio Gagliardi
Attachments: kafka-mm2-checkpoint-race-reproducer-2026-11-06.tar.gz
{{MirrorCheckpointTask.stop()}} closes {{CheckpointStore}} and
{{OffsetSyncStore}} before shutting down the {{Scheduler}} that may still be
asynchronously starting them, so two threads use one {{KafkaConsumer}}
concurrently.
h2. What happens
(Line numbers in this section are from {{trunk}} / 4.4.0; see the per-branch
table under "Affected versions".)
{{MirrorCheckpointTask.start()}} hands the long-running store initialisation to
a {{Scheduler}} thread so that {{start()}} does not block
({{{}MirrorCheckpointTask.java:116{}}}):
{code:java}
scheduler.executeAsync(() -> {
// loading the stores are potentially long running operations, so they run
asynchronously
// to avoid blocking task::start (until a task has completed starting it
cannot be stopped)
boolean checkpointsReadOk = checkpointStore.start();
offsetSyncStore.start(!checkpointsReadOk);
...
}, "starting checkpoint and offset sync stores");
{code}
{{stop()}} then tears the task down in this order
({{{}MirrorCheckpointTask.java:141-145{}}}):
{code:java}
public void stop() {
stopping = true;
Utils.closeQuietly(topicFilter, "topic filter");
Utils.closeQuietly(checkpointStore, "checkpoints store");
Utils.closeQuietly(offsetSyncStore, "offset sync store"); // line 141
...
Utils.closeQuietly(scheduler, "scheduler"); // line 145
}
{code}
The {{offsetSyncStore}} is closed at line 141, but the only thing that waits
for the async initialiser — {{{}Scheduler.close(){}}}, which calls
{{executor.shutdown()}} and {{awaitTermination(...)}} — does not run until line
145. If a rebalance stops the task while the async init is still in flight, the
*task thread* closes the {{KafkaConsumer}} that the *Scheduler thread* is
actively using. {{KafkaConsumer}} is explicitly not safe for multi-threaded
access, so whichever thread calls {{acquire()}} second throws
{{{}ConcurrentModificationException{}}}.
h2. CheckpointStore has the same race
The observed stack traces are all from {{{}OffsetSyncStore{}}}, but
{{checkpointStore.start()}} runs on the same scheduler thread, and
{{closeQuietly(checkpointStore, ...)}} runs on the task thread one line earlier
than the offset sync store. {{CheckpointStore}} owns a second, independent
{{{}KafkaConsumer{}}}, and closes it by the same route ({{{}close(){}}} →
{{releaseResources()}} → {{backingStore::stop}} → {{KafkaBasedLog.stop()}} →
{{{}consumer.close(){}}}). A task stopped while {{readCheckpointsImpl()}} is
inside {{backingStore.start(true)}} races on the checkpoints-topic consumer
exactly as it does on the offset-syncs one.
Two aggravating details specific to this store:
* {{releaseResources()}} is reachable concurrently from both threads — the
scheduler thread runs it in a {{finally}} block, and the task thread runs it
via {{close()}} — so {{backingStore}} and {{cpAdmin}} can be closed twice, from
two threads.
* {{backingStore}} and {{cpAdmin}} are plain (non-{{{}volatile{}}}) fields
written on the scheduler thread and read on the task thread, so there is no
guaranteed visibility of the consumer the task thread is trying to close.
h2. Why KafkaBasedLog.stop() does not protect against this
{{KafkaBasedLog.stop()}} looks like it waits for its reader before closing the
consumer, but the guard does not apply during startup:
{code:java}
public void stop() {
synchronized (this) { stopRequested = true; }
if (consumer != null) consumer.wakeup();
if (thread != null) { thread.join(); } // thread is still null here
...
Utils.closeQuietly(consumer, "KafkaBasedLog consumer for topic " + topic);
}
{code}
{{thread}} (the {{{}WorkThread{}}}) is only assigned *after* the blocking part
of {{start()}} has completed. While {{start()}} is still running, {{{}thread ==
null{}}}, the {{join()}} is skipped, and {{stop()}} proceeds directly to
{{consumer.close()}} — concurrently with the Scheduler thread's use of that
same consumer inside {{{}start(){}}}. Note also that {{start()}} never checks
the {{stopRequested}} flag that {{stop()}} sets.
h2. Observed manifestations
Both are the same race, caught at different instants.
*1. {{wakeup()}} lands while the init thread is blocked in {{poll()}}* — the
reported production symptom, a pair of exceptions:
{code:java}
ERROR [Scheduler for MirrorCheckpointTask:
A->B|A->B.MirrorCheckpointConnector-3-starting checkpoint and offset sync
stores]
Scheduler - ... caught exception in scheduled task: starting checkpoint
and offset sync stores
org.apache.kafka.common.errors.WakeupException: null
at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeTriggerWakeup(ConsumerNetworkClient.java:531)
at
org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.poll(ClassicKafkaConsumer.java:641)
at
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:895)
at
org.apache.kafka.connect.util.KafkaBasedLog.poll(KafkaBasedLog.java:443)
at
org.apache.kafka.connect.util.KafkaBasedLog.readToLogEnd(KafkaBasedLog.java:482)
at
org.apache.kafka.connect.util.KafkaBasedLog.start(KafkaBasedLog.java:278)
at
org.apache.kafka.connect.mirror.OffsetSyncStore.backingStoreStart(OffsetSyncStore.java:121)
at
org.apache.kafka.connect.mirror.OffsetSyncStore.start(OffsetSyncStore.java:115)
at
org.apache.kafka.connect.mirror.MirrorCheckpointTask.lambda$start$1(MirrorCheckpointTask.java:116)
at org.apache.kafka.connect.mirror.Scheduler.run(Scheduler.java:97)
WARN [task-thread-A->B.MirrorCheckpointConnector-3]
Utils - Failed to close KafkaBasedLog consumer for topic
mm2-offset-syncs.A.internal
java.util.ConcurrentModificationException: KafkaConsumer is not safe for
multi-threaded access.
currentThread(name: task-thread-A->B.MirrorCheckpointConnector-3, id: 72)
otherThread(id: 143)
at
org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.acquire(ClassicKafkaConsumer.java:1256)
at
org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.close(ClassicKafkaConsumer.java:1127)
at
org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:1782)
{code}
Here {{WakeupException}} is the documented, expected effect of
{{{}wakeup(){}}}; the {{ConcurrentModificationException}} on {{close()}} is the
defect.
*2. The close lands while the init thread is in {{partitionsFor()}}* — the
reproducer hits this variant most often. The
{{ConcurrentModificationException}} surfaces on the Scheduler thread instead,
with no {{WakeupException}} at all:
{code:java}
ERROR Scheduler for MirrorCheckpointTask: dr->local|MirrorCheckpointConnector-5
caught exception in scheduled task: starting checkpoint and offset sync
stores
java.util.ConcurrentModificationException: KafkaConsumer is not safe for
multi-threaded access.
currentThread(name: Scheduler for MirrorCheckpointTask:
dr->local|MirrorCheckpointConnector-5-starting
checkpoint and offset sync stores, id: 1640) otherThread(id: 295)
at
org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.acquire(ClassicKafkaConsumer.java:1256)
at
org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.partitionsFor(ClassicKafkaConsumer.java:931)
at
org.apache.kafka.clients.consumer.KafkaConsumer.partitionsFor(KafkaConsumer.java:1418)
at
org.apache.kafka.connect.util.KafkaBasedLog.start(KafkaBasedLog.java:248)
at
org.apache.kafka.connect.mirror.OffsetSyncStore.backingStoreStart(OffsetSyncStore.java:121)
at
org.apache.kafka.connect.mirror.OffsetSyncStore.start(OffsetSyncStore.java:115)
at
org.apache.kafka.connect.mirror.MirrorCheckpointTask.lambda$start$1(MirrorCheckpointTask.java:116)
at org.apache.kafka.connect.mirror.Scheduler.run(Scheduler.java:97)
{code}
h2. Trigger
Anything that makes {{MirrorCheckpointConnector}} call
{{context.requestTaskReconfiguration()}} often enough to stop checkpoint tasks
shortly after they start. In the field this was a short
{{refresh.groups.interval.seconds}} combined with a broad {{groups}} pattern
matching churny consumer-group IDs: each change to the matched group set emits
{{Tasks [...] configs updated}} and rebalances the checkpoint tasks.
h2. Impact
* Recurring {{{}ERROR{}}}/{{{}WARN{}}} with stack traces on every affected
rebalance. The task itself is not killed and mirroring continues, so the
primary cost is alert fatigue and the difficulty of distinguishing this from a
real fault.
* *Leaked consumer — confirmed.* In manifestation (1) the
{{ConcurrentModificationException}} is thrown from inside
{{{}KafkaConsumer.close(){}}}, called through {{{}Utils.closeQuietly{}}}, which
swallows it. {{close()}} therefore never completes and the consumer is left
open, leaking its sockets and metrics once per affected task stop.
h2. Reproduction
Reproduced on 4.1.0. A standalone container-based reproducer is attached — two
single-node KRaft clusters plus an MM2 driver, all running the stock
{{apache/kafka:4.1.0}} image, driven by a single {{{}run.sh{}}}; podman or
docker is the only prerequisite. Steps:
# Start two single-node KRaft clusters, {{dr}} (source) and {{local}} (target).
# Create a few topics on {{dr}} matching the mirrored pattern and seed them
with a few thousand records (widens the race window).
# Run {{connect-mirror-maker.sh}} for {{dr->local}} with {{{}groups =
^(.{*}prod.{*})${}}}, {{{}refresh.groups.interval.seconds = 5{}}},
{{{}emit.checkpoints.enabled = true{}}}, and {{{}tasks.max = 8{}}}.
# Continuously create new consumer groups on {{dr}} matching that pattern (one
every ~3s), each committing offsets.
# Grep the MM2 driver log for {{ConcurrentModificationException}} /
{{{}WakeupException{}}}.
{*}Observed{*}: on the attached reproducer, 5 hits over a ~5 minute run — four
different checkpoint tasks, distinct thread IDs, all the manifestation (2)
variant above. An earlier run on a different 4.1.0 build saw 1 hit over ~5
minutes and 36 {{Tasks [...] configs updated}} events. The rate is
timing-dependent and varies with host speed, so a clean run is not evidence of
absence; re-run or raise the churn rate.
h2. Affected versions
*Reproduced* on 4.1.0. Still present on {{trunk}} (commit {{{}98e0bebf80{}}})
and on 4.4.0, the latest release. The racing code is substantively unchanged
from 4.1 through {{{}trunk{}}}.
||Branch||{{executeAsync}}||{{closeQuietly(offsetSyncStore)}}||{{closeQuietly(scheduler)}}||
|3.8|— (blocking {{{}execute{}}})|137|141|
|3.9|112|137|141|
|4.0|113|138|142|
|4.1|112|137|141|
|4.2|111|136|140|
|4.3|116|141|145|
|4.4|116|141|145|
|trunk|116|141|145|
The race became reachable on the normal code path in 3.9.0, via KAFKA-17242 /
commit {{{}4763d0fc30cd{}}}, which replaced the blocking
{{scheduler.execute(...)}} with {{{}scheduler.executeAsync(...){}}}:
* {*}3.7, 3.8 and earlier{*}: {{{}scheduler.execute(...){}}}, which blocks on
{{{}Future.get(admin.timeout.ms){}}}. {{start()}} normally returns only after
the stores are initialised, so the task thread normally cannot reach {{stop()}}
mid-init. However, {{execute()}} only waits up to the timeout — on timeout it
returns while the submitted task keeps running. So the same collision was
possible on 3.8 and earlier, but only when store initialisation overran
{{{}admin.timeout.ms{}}}.
* {*}3.9.0 onward{*}: {{executeAsync(...)}} returns immediately by design, so
every task start leaves an initialisation running concurrently with a task that
is now eligible to be stopped. What changed in 3.9.0 is that the overlap went
from an edge case to the normal path.
h2. Relationship to existing issues
* *KAFKA-13856* ("MirrorCheckpointTask meets ConcurrentModificationException",
still Open, affects 3.0.1) — *not a duplicate.* That report is about the
{{checkpointsPerConsumerGroup}} {{HashMap}} field being mutated and traversed
by the scheduler and worker threads concurrently. Same exception type and same
class, different object and different fix. This issue is about the
{{KafkaConsumer}} inside {{{}OffsetSyncStore{}}}'s {{{}KafkaBasedLog{}}}, and
its stack traces go through {{{}ClassicKafkaConsumer.acquire{}}}, not
{{{}HashMap{}}}.
* *KAFKA-17242* (Fixed, 3.9.0) — the change that introduced the async start.
Not a regression to revert: reverting it would restore the spurious timeout it
fixed. This issue should be fixed on its own terms.
h2. Proposed fix
A candidate patch against {{trunk}} is ready (6 files, +352/−8, confined to
{{{}connect/mirror{}}}; no public API change, so no KIP). {{stop()}} closes the
{{Scheduler}} before the stores, so {{Scheduler.close()}} returns only after
{{awaitTermination()}} succeeds and the stores are closed strictly after the
thread loading them has finished. To bound that wait, each store gains a
{{stopInitialization()}} method, called first, which sets a flag and calls
{{Consumer::wakeup}} — the one consumer method safe from another thread. The
resulting {{WakeupException}} is logged at DEBUG as an orderly shutdown rather
than ERROR. {{start()}} remains non-blocking, so KAFKA-17242 is not regressed.
h3. Known limitation
{{{}Scheduler.close(){}}}'s {{awaitTermination}} timeout remains; if it
expires, {{close()}} logs and returns, {{stop()}} proceeds, and the race window
reopens as before. This is pre-existing behaviour shared with the other MM2
tasks. {*}Fixed except for a pre-existing escape hatch{*}, not {*}proven
impossible{*}.
h3. Rejected alternatives
* *Guard the stores with a shutdown flag / make {{close()}} safe* — a flag
gives no happens-before edge; doing it properly means holding a lock across the
whole of {{{}KafkaBasedLog.start(){}}}, which reinstates the unbounded wait.
* *Fix {{KafkaBasedLog}} so {{stop()}} waits for an in-progress {{start()}}* —
the most general fix, but {{KafkaBasedLog}} is shared with the Connect herder
where {{{}start(){}}}/{{{}stop(){}}} are single-threaded, so this is risk with
no local benefit. Worth a separate hardening JIRA.
* *Revert to the blocking {{scheduler.execute(...)}}* — reintroduces
KAFKA-17242's spurious timeout and makes {{start()}} block again.
* *{{shutdownNow()}} / interrupting the scheduler thread* — turns the blocked
{{poll()}} into an {{InterruptException}} and leaves the consumer in an
undefined state; {{wakeup()}} is the supported mechanism and leaves it
closeable.
h3. Tests
{{MirrorCheckpointTaskTest.testStopDoesNotCloseOffsetSyncStoreWhileItIsStarting}}
and {{...testStopDoesNotCloseCheckpointStoreWhileItIsStarting}} block a store
mid-startup on the scheduler thread, call {{stop()}} from the task thread, and
assert the store was not closed while that startup was in progress. The shipped
race tests assert the *ordering invariant* using fakes, which is what makes
them deterministic; the link from "ordering violated" to
{{ConcurrentModificationException}} is established separately by the production
stack traces above and by a single-threaded-access-checking consumer driven
through a real {{KafkaBasedLog}} against unmodified {{{}trunk{}}}.
h2. Workaround
None that removes the race. Increasing {{refresh.groups.interval.seconds}} and
narrowing the {{groups}} pattern reduces checkpoint-task rebalances and
therefore how often it fires. The exceptions are logged and swallowed, so
mirroring is not interrupted.
Assisted by Claude Opus 5
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
