[
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 possibly 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.)
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 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.
> 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.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 possibly 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.)
> 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)
