[
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 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.
was:
After a StreamThread dies and releases its task locks, the background
state-directory cleaner can delete a still-valid task directory whose task
committed only seconds earlier. The reassigned task then finds no local
offsets, restores its changelog from the beginning, and — for a
compacted/retention-deleted changelog — possibly the restore is overtaken by
retention mid-flight, producing \{{OffsetOutOfRangeException}} →
\{{TaskCorruptedException}} → rebalance.
The cleaner decides obsolescence from the task directory's filesystem mtime:
\{{now - taskDir.lastModified() > state.cleanup.delay.ms}}
(StateDirectory.cleanRemovedTasks). This relied on the pre-KIP-1035 behavior
where the per-commit \{{.checkpoint}} file (a direct child of the task dir)
refreshed the dir mtime on every commit.
> 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 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.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
