[jira] [Commented] (KAFKA-8574) EOS race condition during task transition leads to LocalStateStore truncation in Kafka Streams 2.0.1
[ https://issues.apache.org/jira/browse/KAFKA-8574?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17030829#comment-17030829 ] Guozhang Wang commented on KAFKA-8574: -- As part of KAFKA-9113 we have moved the reading of checkpoint file out of the ProcessorStateManager's constructor, so I think we can resolve this ticket as a piggy-backed fix along with 9113. > EOS race condition during task transition leads to LocalStateStore truncation > in Kafka Streams 2.0.1 > > > Key: KAFKA-8574 > URL: https://issues.apache.org/jira/browse/KAFKA-8574 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.0.1 >Reporter: William Greer >Priority: Major > > *Overview* > While using EOS in Kafka Stream there is a race condition where the > checkpoint file is written by the previous owning thread (Thread A) after the > new owning thread (Thread B) reads the checkpoint file. Thread B then starts > a restoration since no checkpoint file was found. A re-balance occurs before > Thread B completes the restoration and a third Thread (Thread C) becomes the > owning thread (Thread C) reads the checkpoint file written by Thread A which > does not correspond to the current state of the RocksDB state store. When > this race condition occurs the state store will have the most recent records > and some amount of the oldest records but will be missing some amount of > records in between. If A->Z represents the entire changelog to the present > then when this scenario occurs the state store would contain records [A->K > and Y->Z] where the state store is missing records K->Y. > > This race condition is possible due to dirty writes and dirty reads of the > checkpoint file. > > *Example:* > Thread refers to a Kafka Streams StreamThread [0] > Thread A, B and C are running in the same JVM in the same streams > application. > > Scenario: > Thread-A is in RUNNING state and up to date on partition 1. > Thread-A is suspended on 1. This does not write a checkpoint file because > EOS is enabled [1] > Thread-B is assigned to 1 > Thread-B does not find checkpoint in StateManager [2] > Thread-A is assigned a different partition. Task writes suspended tasks > checkpoints to disk. Checkpoint for 1 is written. [3] > Thread-B deletes LocalStore and starts restoring. The deletion of the > LocalStore does not delete checkpoint file. [4] > Thread-C is revoked > Thread-A is revoked > Thread-B is revoked from the assigned status. Does not write a checkpoint > file > - Note Thread-B never reaches the running state, it remains in the > PARTITIONS_ASSIGNED state until it transitions to the PARTITIONS_REVOKED state > Thread-C is assigned 1 > Thread-C finds checkpoint in StateManager. This checkpoint corresponds to > where Thread-A left the state store for partition 1 at and not where Thread-B > left the state store at. > Thread-C begins restoring from checkpoint. The state store is missing an > unknown number of records at this point > Thread-B is assigned does not write a checkpoint file for partition 1, > because it had not reached a running status before being revoked > > [0] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java] > [1] > [https://github.com/apache/kafka/blob/2.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L522-L553] > [2] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java#L98] > [3] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L104-L105] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java#L316-L331] > [4] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java#L228] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java#L62-L123] > Specifically > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java#L107-L119] > is where the state store is deleted but the checkpoint file is not. > > *How we recovered:* > 1. Deleted the impacted state store. This triggered multiple exceptions and > initiated a re-balance. > > *Possible approaches to address this issue:* > 1. Add a collection of global task locks for concurrency protection of the > checkpoint file. With the lock for
[jira] [Commented] (KAFKA-8574) EOS race condition during task transition leads to LocalStateStore truncation in Kafka Streams 2.0.1
[ https://issues.apache.org/jira/browse/KAFKA-8574?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17005864#comment-17005864 ] Sophie Blee-Goldman commented on KAFKA-8574: Given the fix [~guozhang] mentioned above, this race condition is now extremely unlikely, but I think it's still possible if suspension takes a very long time (FWIW KIP-429 should make it even less likely in 2.4) IMO we should move the checkpoint file read out of the ProcessorStateManager's constructor so that the task can at least grab the state directory lock before doing this kind of initialization. > EOS race condition during task transition leads to LocalStateStore truncation > in Kafka Streams 2.0.1 > > > Key: KAFKA-8574 > URL: https://issues.apache.org/jira/browse/KAFKA-8574 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.0.1 >Reporter: William Greer >Priority: Major > > *Overview* > While using EOS in Kafka Stream there is a race condition where the > checkpoint file is written by the previous owning thread (Thread A) after the > new owning thread (Thread B) reads the checkpoint file. Thread B then starts > a restoration since no checkpoint file was found. A re-balance occurs before > Thread B completes the restoration and a third Thread (Thread C) becomes the > owning thread (Thread C) reads the checkpoint file written by Thread A which > does not correspond to the current state of the RocksDB state store. When > this race condition occurs the state store will have the most recent records > and some amount of the oldest records but will be missing some amount of > records in between. If A->Z represents the entire changelog to the present > then when this scenario occurs the state store would contain records [A->K > and Y->Z] where the state store is missing records K->Y. > > This race condition is possible due to dirty writes and dirty reads of the > checkpoint file. > > *Example:* > Thread refers to a Kafka Streams StreamThread [0] > Thread A, B and C are running in the same JVM in the same streams > application. > > Scenario: > Thread-A is in RUNNING state and up to date on partition 1. > Thread-A is suspended on 1. This does not write a checkpoint file because > EOS is enabled [1] > Thread-B is assigned to 1 > Thread-B does not find checkpoint in StateManager [2] > Thread-A is assigned a different partition. Task writes suspended tasks > checkpoints to disk. Checkpoint for 1 is written. [3] > Thread-B deletes LocalStore and starts restoring. The deletion of the > LocalStore does not delete checkpoint file. [4] > Thread-C is revoked > Thread-A is revoked > Thread-B is revoked from the assigned status. Does not write a checkpoint > file > - Note Thread-B never reaches the running state, it remains in the > PARTITIONS_ASSIGNED state until it transitions to the PARTITIONS_REVOKED state > Thread-C is assigned 1 > Thread-C finds checkpoint in StateManager. This checkpoint corresponds to > where Thread-A left the state store for partition 1 at and not where Thread-B > left the state store at. > Thread-C begins restoring from checkpoint. The state store is missing an > unknown number of records at this point > Thread-B is assigned does not write a checkpoint file for partition 1, > because it had not reached a running status before being revoked > > [0] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java] > [1] > [https://github.com/apache/kafka/blob/2.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L522-L553] > [2] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java#L98] > [3] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L104-L105] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java#L316-L331] > [4] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java#L228] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java#L62-L123] > Specifically > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java#L107-L119] > is where the state store is deleted but the checkpoint file is not. > > *How we recovered:* > 1. Deleted the impacted state store. This
[jira] [Commented] (KAFKA-8574) EOS race condition during task transition leads to LocalStateStore truncation in Kafka Streams 2.0.1
[ https://issues.apache.org/jira/browse/KAFKA-8574?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16923831#comment-16923831 ] Guozhang Wang commented on KAFKA-8574: -- Sorry for being late on this ticket. I've read through the code and I agree with [~wgreerx]'s analysis that this is indeed a bug. As for the fix, I think this PR which is merged in 2.3.0 / 2.2.1 should have resolved it since it now writes a checkpoint file upon suspended with EOS turned on still, and hence we tackle this issue by fixing the second step: "Thread-A is suspended on 1. This does not write a checkpoint file because EOS is enabled [1]". https://github.com/apache/kafka/commit/1f9aa01a5b3b59d90499a059d719af03483d5130 > EOS race condition during task transition leads to LocalStateStore truncation > in Kafka Streams 2.0.1 > > > Key: KAFKA-8574 > URL: https://issues.apache.org/jira/browse/KAFKA-8574 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.0.1 >Reporter: William Greer >Priority: Major > > *Overview* > While using EOS in Kafka Stream there is a race condition where the > checkpoint file is written by the previous owning thread (Thread A) after the > new owning thread (Thread B) reads the checkpoint file. Thread B then starts > a restoration since no checkpoint file was found. A re-balance occurs before > Thread B completes the restoration and a third Thread (Thread C) becomes the > owning thread (Thread C) reads the checkpoint file written by Thread A which > does not correspond to the current state of the RocksDB state store. When > this race condition occurs the state store will have the most recent records > and some amount of the oldest records but will be missing some amount of > records in between. If A->Z represents the entire changelog to the present > then when this scenario occurs the state store would contain records [A->K > and Y->Z] where the state store is missing records K->Y. > > This race condition is possible due to dirty writes and dirty reads of the > checkpoint file. > > *Example:* > Thread refers to a Kafka Streams StreamThread [0] > Thread A, B and C are running in the same JVM in the same streams > application. > > Scenario: > Thread-A is in RUNNING state and up to date on partition 1. > Thread-A is suspended on 1. This does not write a checkpoint file because > EOS is enabled [1] > Thread-B is assigned to 1 > Thread-B does not find checkpoint in StateManager [2] > Thread-A is assigned a different partition. Task writes suspended tasks > checkpoints to disk. Checkpoint for 1 is written. [3] > Thread-B deletes LocalStore and starts restoring. The deletion of the > LocalStore does not delete checkpoint file. [4] > Thread-C is revoked > Thread-A is revoked > Thread-B is revoked from the assigned status. Does not write a checkpoint > file > - Note Thread-B never reaches the running state, it remains in the > PARTITIONS_ASSIGNED state until it transitions to the PARTITIONS_REVOKED state > Thread-C is assigned 1 > Thread-C finds checkpoint in StateManager. This checkpoint corresponds to > where Thread-A left the state store for partition 1 at and not where Thread-B > left the state store at. > Thread-C begins restoring from checkpoint. The state store is missing an > unknown number of records at this point > Thread-B is assigned does not write a checkpoint file for partition 1, > because it had not reached a running status before being revoked > > [0] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java] > [1] > [https://github.com/apache/kafka/blob/2.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L522-L553] > [2] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java#L98] > [3] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L104-L105] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java#L316-L331] > [4] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java#L228] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java#L62-L123] > Specifically > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java#L107-L119] > is where the state store is
[jira] [Commented] (KAFKA-8574) EOS race condition during task transition leads to LocalStateStore truncation in Kafka Streams 2.0.1
[ https://issues.apache.org/jira/browse/KAFKA-8574?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16891259#comment-16891259 ] Matthias J. Sax commented on KAFKA-8574: I cannot remember all the details. What was the deadlock issue? We should hand-over tasks from one thread to another within an instance using the .lock files as synchronization points. Another high level idea would be, to actually write the checkpoint file on suspend() (instead of close()) and delete it on resume()? \cc [~guozhang] > EOS race condition during task transition leads to LocalStateStore truncation > in Kafka Streams 2.0.1 > > > Key: KAFKA-8574 > URL: https://issues.apache.org/jira/browse/KAFKA-8574 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.0.1 >Reporter: William Greer >Priority: Major > > *Overview* > While using EOS in Kafka Stream there is a race condition where the > checkpoint file is written by the previous owning thread (Thread A) after the > new owning thread (Thread B) reads the checkpoint file. Thread B then starts > a restoration since no checkpoint file was found. A re-balance occurs before > Thread B completes the restoration and a third Thread (Thread C) becomes the > owning thread (Thread C) reads the checkpoint file written by Thread A which > does not correspond to the current state of the RocksDB state store. When > this race condition occurs the state store will have the most recent records > and some amount of the oldest records but will be missing some amount of > records in between. If A->Z represents the entire changelog to the present > then when this scenario occurs the state store would contain records [A->K > and Y->Z] where the state store is missing records K->Y. > > This race condition is possible due to dirty writes and dirty reads of the > checkpoint file. > > *Example:* > Thread refers to a Kafka Streams StreamThread [0] > Thread A, B and C are running in the same JVM in the same streams > application. > > Scenario: > Thread-A is in RUNNING state and up to date on partition 1. > Thread-A is suspended on 1. This does not write a checkpoint file because > EOS is enabled [1] > Thread-B is assigned to 1 > Thread-B does not find checkpoint in StateManager [2] > Thread-A is assigned a different partition. Task writes suspended tasks > checkpoints to disk. Checkpoint for 1 is written. [3] > Thread-B deletes LocalStore and starts restoring. The deletion of the > LocalStore does not delete checkpoint file. [4] > Thread-C is revoked > Thread-A is revoked > Thread-B is revoked from the assigned status. Does not write a checkpoint > file > - Note Thread-B never reaches the running state, it remains in the > PARTITIONS_ASSIGNED state until it transitions to the PARTITIONS_REVOKED state > Thread-C is assigned 1 > Thread-C finds checkpoint in StateManager. This checkpoint corresponds to > where Thread-A left the state store for partition 1 at and not where Thread-B > left the state store at. > Thread-C begins restoring from checkpoint. The state store is missing an > unknown number of records at this point > Thread-B is assigned does not write a checkpoint file for partition 1, > because it had not reached a running status before being revoked > > [0] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java] > [1] > [https://github.com/apache/kafka/blob/2.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L522-L553] > [2] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java#L98] > [3] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L104-L105] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java#L316-L331] > [4] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java#L228] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java#L62-L123] > Specifically > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java#L107-L119] > is where the state store is deleted but the checkpoint file is not. > > *How we recovered:* > 1. Deleted the impacted state store. This triggered multiple exceptions and > initiated a re-balance. > > *Possible
[jira] [Commented] (KAFKA-8574) EOS race condition during task transition leads to LocalStateStore truncation in Kafka Streams 2.0.1
[ https://issues.apache.org/jira/browse/KAFKA-8574?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16871837#comment-16871837 ] William Greer commented on KAFKA-8574: -- Looking at the code more I think the race condition could be addressed by acquiring the state directory lock before reading the checkpoint file in the ProcessorStateManager. Looking at the code history shows that there used to be a lock acquisition before the checkpoint read but it was removed in KAFKA-5152 to address a deadlock during restoration in 0.10.2.1. Does adding back the lock acquisition before reading the checkpoint file re-introduce a deadlock issue? > EOS race condition during task transition leads to LocalStateStore truncation > in Kafka Streams 2.0.1 > > > Key: KAFKA-8574 > URL: https://issues.apache.org/jira/browse/KAFKA-8574 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.0.1 >Reporter: William Greer >Priority: Major > > *Overview* > While using EOS in Kafka Stream there is a race condition where the > checkpoint file is written by the previous owning thread (Thread A) after the > new owning thread (Thread B) reads the checkpoint file. Thread B then starts > a restoration since no checkpoint file was found. A re-balance occurs before > Thread B completes the restoration and a third Thread (Thread C) becomes the > owning thread (Thread C) reads the checkpoint file written by Thread A which > does not correspond to the current state of the RocksDB state store. When > this race condition occurs the state store will have the most recent records > and some amount of the oldest records but will be missing some amount of > records in between. If A->Z represents the entire changelog to the present > then when this scenario occurs the state store would contain records [A->K > and Y->Z] where the state store is missing records K->Y. > > This race condition is possible due to dirty writes and dirty reads of the > checkpoint file. > > *Example:* > Thread refers to a Kafka Streams StreamThread [0] > Thread A, B and C are running in the same JVM in the same streams > application. > > Scenario: > Thread-A is in RUNNING state and up to date on partition 1. > Thread-A is suspended on 1. This does not write a checkpoint file because > EOS is enabled [1] > Thread-B is assigned to 1 > Thread-B does not find checkpoint in StateManager [2] > Thread-A is assigned a different partition. Task writes suspended tasks > checkpoints to disk. Checkpoint for 1 is written. [3] > Thread-B deletes LocalStore and starts restoring. The deletion of the > LocalStore does not delete checkpoint file. [4] > Thread-C is revoked > Thread-A is revoked > Thread-B is revoked from the assigned status. Does not write a checkpoint > file > - Note Thread-B never reaches the running state, it remains in the > PARTITIONS_ASSIGNED state until it transitions to the PARTITIONS_REVOKED state > Thread-C is assigned 1 > Thread-C finds checkpoint in StateManager. This checkpoint corresponds to > where Thread-A left the state store for partition 1 at and not where Thread-B > left the state store at. > Thread-C begins restoring from checkpoint. The state store is missing an > unknown number of records at this point > Thread-B is assigned does not write a checkpoint file for partition 1, > because it had not reached a running status before being revoked > > [0] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java] > [1] > [https://github.com/apache/kafka/blob/2.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L522-L553] > [2] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java#L98] > [3] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L104-L105] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java#L316-L331] > [4] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java#L228] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java#L62-L123] > Specifically > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java#L107-L119] > is where the state store is deleted but the checkpoint file is not. > > *How we
[jira] [Commented] (KAFKA-8574) EOS race condition during task transition leads to LocalStateStore truncation in Kafka Streams 2.0.1
[ https://issues.apache.org/jira/browse/KAFKA-8574?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16869529#comment-16869529 ] William Greer commented on KAFKA-8574: -- This is a different issue from KAFKA-8187. KAFKA-8187 had the possible local data loss when standby tasks were in use. This issue can have possible data loss regardless of usage of standbys as long as EOS is enabled. This issue may be mitigated by KAFKA-7672 but the race condition that makes this possible still exists even with the KAFKA-7672 change-set applied. The race conditions around the unprotected access of the checkpoint files (Writes and reads of the checkpoint without a lock) still exist in trunk, whether there are any correctness issues when the race condition occurs is a different question. > EOS race condition during task transition leads to LocalStateStore truncation > in Kafka Streams 2.0.1 > > > Key: KAFKA-8574 > URL: https://issues.apache.org/jira/browse/KAFKA-8574 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.0.1 >Reporter: William Greer >Priority: Major > > *Overview* > While using EOS in Kafka Stream there is a race condition where the > checkpoint file is written by the previous owning thread (Thread A) after the > new owning thread (Thread B) reads the checkpoint file. Thread B then starts > a restoration since no checkpoint file was found. A re-balance occurs before > Thread B completes the restoration and a third Thread (Thread C) becomes the > owning thread (Thread C) reads the checkpoint file written by Thread A which > does not correspond to the current state of the RocksDB state store. When > this race condition occurs the state store will have the most recent records > and some amount of the oldest records but will be missing some amount of > records in between. If A->Z represents the entire changelog to the present > then when this scenario occurs the state store would contain records [A->K > and Y->Z] where the state store is missing records K->Y. > > This race condition is possible due to dirty writes and dirty reads of the > checkpoint file. > > *Example:* > Thread refers to a Kafka Streams StreamThread [0] > Thread A, B and C are running in the same JVM in the same streams > application. > > Scenario: > Thread-A is in RUNNING state and up to date on partition 1. > Thread-A is suspended on 1. This does not write a checkpoint file because > EOS is enabled [1] > Thread-B is assigned to 1 > Thread-B does not find checkpoint in StateManager [2] > Thread-A is assigned a different partition. Task writes suspended tasks > checkpoints to disk. Checkpoint for 1 is written. [3] > Thread-B deletes LocalStore and starts restoring. The deletion of the > LocalStore does not delete checkpoint file. [4] > Thread-C is revoked > Thread-A is revoked > Thread-B is revoked from the assigned status. Does not write a checkpoint > file > - Note Thread-B never reaches the running state, it remains in the > PARTITIONS_ASSIGNED state until it transitions to the PARTITIONS_REVOKED state > Thread-C is assigned 1 > Thread-C finds checkpoint in StateManager. This checkpoint corresponds to > where Thread-A left the state store for partition 1 at and not where Thread-B > left the state store at. > Thread-C begins restoring from checkpoint. The state store is missing an > unknown number of records at this point > Thread-B is assigned does not write a checkpoint file for partition 1, > because it had not reached a running status before being revoked > > [0] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java] > [1] > [https://github.com/apache/kafka/blob/2.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L522-L553] > [2] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java#L98] > [3] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L104-L105] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java#L316-L331] > [4] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java#L228] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java#L62-L123] > Specifically >
[jira] [Commented] (KAFKA-8574) EOS race condition during task transition leads to LocalStateStore truncation in Kafka Streams 2.0.1
[ https://issues.apache.org/jira/browse/KAFKA-8574?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16869051#comment-16869051 ] Matthias J. Sax commented on KAFKA-8574: Is this the same as https://issues.apache.org/jira/browse/KAFKA-8187 or different? > EOS race condition during task transition leads to LocalStateStore truncation > in Kafka Streams 2.0.1 > > > Key: KAFKA-8574 > URL: https://issues.apache.org/jira/browse/KAFKA-8574 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.0.1 >Reporter: William Greer >Priority: Major > > *Overview* > While using EOS in Kafka Stream there is a race condition where the > checkpoint file is written by the previous owning thread (Thread A) after the > new owning thread (Thread B) reads the checkpoint file. Thread B then starts > a restoration since no checkpoint file was found. A re-balance occurs before > Thread B completes the restoration and a third Thread (Thread C) becomes the > owning thread (Thread C) reads the checkpoint file written by Thread A which > does not correspond to the current state of the RocksDB state store. When > this race condition occurs the state store will have the most recent records > and some amount of the oldest records but will be missing some amount of > records in between. If A->Z represents the entire changelog to the present > then when this scenario occurs the state store would contain records [A->K > and Y->Z] where the state store is missing records K->Y. > > This race condition is possible due to dirty writes and dirty reads of the > checkpoint file. > > *Example:* > Thread refers to a Kafka Streams StreamThread [0] > Thread A, B and C are running in the same JVM in the same streams > application. > > Scenario: > Thread-A is in RUNNING state and up to date on partition 1. > Thread-A is suspended on 1. This does not write a checkpoint file because > EOS is enabled [1] > Thread-B is assigned to 1 > Thread-B does not find checkpoint in StateManager [2] > Thread-A is assigned a different partition. Task writes suspended tasks > checkpoints to disk. Checkpoint for 1 is written. [3] > Thread-B deletes LocalStore and starts restoring. The deletion of the > LocalStore does not delete checkpoint file. [4] > Thread-C is revoked > Thread-A is revoked > Thread-B is revoked from the assigned status. Does not write a checkpoint > file > - Note Thread-B never reaches the running state, it remains in the > PARTITIONS_ASSIGNED state until it transitions to the PARTITIONS_REVOKED state > Thread-C is assigned 1 > Thread-C finds checkpoint in StateManager. This checkpoint corresponds to > where Thread-A left the state store for partition 1 at and not where Thread-B > left the state store at. > Thread-C begins restoring from checkpoint. The state store is missing an > unknown number of records at this point > Thread-B is assigned does not write a checkpoint file for partition 1, > because it had not reached a running status before being revoked > > [0] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java] > [1] > [https://github.com/apache/kafka/blob/2.0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L522-L553] > [2] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java#L98] > [3] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L104-L105] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedTasks.java#L316-L331] > [4] > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java#L228] > & > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java#L62-L123] > Specifically > [https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractStateManager.java#L107-L119] > is where the state store is deleted but the checkpoint file is not. > > *How we recovered:* > 1. Deleted the impacted state store. This triggered multiple exceptions and > initiated a re-balance. > > *Possible approaches to address this issue:* > 1. Add a collection of global task locks for concurrency protection of the > checkpoint file. With the lock for suspended tasks being released after > closeNonAssignedSuspendedTasks and the locks being acquired after
