[
https://issues.apache.org/jira/browse/KAFKA-20805?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Bill Bejeck updated KAFKA-20805:
--------------------------------
Description:
After a StreamThread failure (thread replaced, task reassigned to another
thread on the same instance), the background cleaner
(\{{StateDirectory.cleanRemovedTasks}}) can delete the task's still-valid,
recently-committed state directory in the window between the failed thread
releasing its lock and the new thread re-acquiring it. The reassigned task then
finds no local offsets and restores its entire
changelog from scratch; for a compacted changelog whose head is being removed
by retention, the from-beginning restore can be overtaken mid-flight →
\{{OffsetOutOfRangeException}} →
\{{TaskCorruptedException}}.
Applies when the dirty close does not wipe local state — i.e. ALOS, or EOS
with transactional state stores (KIP-892). (Under EOS with non-transactional
stores the dirty close wipes at close, so this path isn't hit.)
Root cause is KIP-1035. The cleaner judges obsolescence by \{{now -
taskDir.lastModified() > state.cleanup.delay.ms}}. Pre-KIP-1035 the per-commit
\{{.checkpoint}} file (a direct child of the task dir,
written by rename) refreshed that mtime on every commit; KIP-1035 moved
offsets into RocksDB and writes \{{.checkpoint}} only on downgrade, so nothing
refreshes the task-dir mtime during processing (RocksDB
writes land in the \{{rocksdb/<store>/}} subdirectory, which doesn't bump the
parent's mtime; per-task locking is in-memory, so there's no \{{.lock}} file
either). The directory looks arbitrarily stale
while its task is actively committing, so the moment its lock is released it
is immediately eligible for deletion instead of after the intended grace period.
was:
After a task's lock is released — via a StreamThread failure, but also a normal
rebalance/migration, partition revocation, or standby→active handoff — the
background state-directory cleaner (\{{StateDirectory.cleanRemovedTasks}}) can
delete a still-valid task directory whose task committed only seconds earlier.
It is therefore not limited to thread failures, and affects both ALOS and EOS
and both transactional and non-transactional stores (not gated on KIP-892).
Root cause is KIP-1035. The cleaner decides obsolescence from the task
directory's mtime: \{{now - taskDir.lastModified() > state.cleanup.delay.ms}}.
Pre-KIP-1035, the per-commit \{{.checkpoint}} file (a
direct child of the task dir, written via rename) refreshed that mtime on every
commit. Under KIP-1035 offsets live in RocksDB and \{{.checkpoint}} is written
only on downgrade, so nothing refreshes the
task-directory mtime during processing — RocksDB writes land in the
\{{rocksdb/<store>/}} subdirectory (which doesn't bump the parent's mtime) and
per-task locking is in-memory (no \{{.lock}} file). The
directory therefore looks arbitrarily stale while its task is actively
committing, so once unlocked it is deleted well before the intended grace
period.
> State directory cleaned prematurely under KIP-1035, forcing from-scratch
> restore
> ---------------------------------------------------------------------------------
>
> Key: KAFKA-20805
> URL: https://issues.apache.org/jira/browse/KAFKA-20805
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 4.3.0, 4.3.1, 4.4.0
> Reporter: Bill Bejeck
> Assignee: Bill Bejeck
> Priority: Blocker
> Fix For: 4.4.0
>
>
> After a StreamThread failure (thread replaced, task reassigned to another
> thread on the same instance), the background cleaner
> (\{{StateDirectory.cleanRemovedTasks}}) can delete the task's still-valid,
>
> recently-committed state directory in the window between the failed thread
> releasing its lock and the new thread re-acquiring it. The reassigned task
> then finds no local offsets and restores its entire
> changelog from scratch; for a compacted changelog whose head is being
> removed by retention, the from-beginning restore can be overtaken mid-flight
> → \{{OffsetOutOfRangeException}} →
> \{{TaskCorruptedException}}.
>
>
>
>
>
> Applies when the dirty close does not wipe local state — i.e. ALOS, or EOS
> with transactional state stores (KIP-892). (Under EOS with non-transactional
> stores the dirty close wipes at close, so this path isn't hit.)
>
>
>
>
>
> Root cause is KIP-1035. The cleaner judges obsolescence by \{{now -
> taskDir.lastModified() > state.cleanup.delay.ms}}. Pre-KIP-1035 the
> per-commit \{{.checkpoint}} file (a direct child of the task dir,
> written by rename) refreshed that mtime on every commit; KIP-1035 moved
> offsets into RocksDB and writes \{{.checkpoint}} only on downgrade, so
> nothing refreshes the task-dir mtime during processing (RocksDB
> writes land in the \{{rocksdb/<store>/}} subdirectory, which doesn't bump
> the parent's mtime; per-task locking is in-memory, so there's no \{{.lock}}
> file either). The directory looks arbitrarily stale
> while its task is actively committing, so the moment its lock is released
> it is immediately eligible for deletion instead of after the intended grace
> period.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
