[jira] [Assigned] (KAFKA-7672) The local state not fully restored after KafkaStream rebalanced, resulting in data loss

["linyue li (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/KAFKA-7672?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

linyue li reassigned KAFKA-7672:
--------------------------------

    Assignee: linyue li

> The local state not fully restored after KafkaStream rebalanced, resulting in data loss
> ---------------------------------------------------------------------------------------
>
>                 Key: KAFKA-7672
>                 URL: https://issues.apache.org/jira/browse/KAFKA-7672
>             Project: Kafka
>          Issue Type: Bug
>          Components: streams
>    Affects Versions: 1.1.0, 1.1.1, 2.0.0, 2.1.0
>            Reporter: linyue li
>            Assignee: linyue li
>            Priority: Major
>             Fix For: 2.1.0
>
>
> Normally, when a task is mitigated to a new thread and no checkpoint file was found under its task folder, Kafka Stream needs to restore the local state for remote changelog topic completely and then resume running. However, in some scenarios, we found that Kafka Stream *NOT* restore this state even no checkpoint was found, but just clean the state folder and transition to running state directly, resulting the historic data loss. 
> To be specific, I will give the detailed logs for Kafka Stream in our project to show this scenario: 
> {quote}2018-10-23 08:27:07,684 INFO  org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=AuditTrailBatch-StreamThread-1-consumer, groupId=AuditTrailBatch] Revoking previously assigned partitions [AuditTrailBatch-0-5]
> 2018-10-23 08:27:07,684 INFO  org.apache.kafka.streams.processor.internals.StreamThread    - stream-thread [AuditTrailBatch-StreamThread-1] State transition from PARTITIONS_ASSIGNED to PARTITIONS_REVOKED
> 2018-10-23 08:27:10,856 INFO  org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=AuditTrailBatch-StreamThread-1-consumer, groupId=AuditTrailBatch] (Re-)joining group
> 2018-10-23 08:27:53,153 INFO  org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=AuditTrailBatch-StreamThread-1-consumer, groupId=AuditTrailBatch] Successfully joined group with generation 323
> 2018-10-23 08:27:53,153 INFO  org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=AuditTrailBatch-StreamThread-1-consumer, groupId=AuditTrailBatch] Setting newly assigned partitions [AuditTrailBatch-store1-repartition-1]
> 2018-10-23 08:27:53,153 INFO  org.apache.kafka.streams.processor.internals.StreamThread    - stream-thread [AuditTrailBatch-StreamThread-1] State transition from PARTITIONS_REVOKED to PARTITIONS_ASSIGNED
> 2018-10-23 08:27:53,153 INFO  org.apache.kafka.streams.processor.internals.StreamThread    - stream-thread [AuditTrailBatch-StreamThread-1] *Creating producer client for task 1_1*
> 2018-10-23 08:27:53,622 INFO  org.apache.kafka.streams.processor.internals.StreamThread    - stream-thread [AuditTrailBatch-StreamThread-1] partition assignment took 469 ms.
> 2018-10-23 08:27:54,357 INFO  org.apache.kafka.streams.processor.internals.StoreChangelogReader - stream-thread [AuditTrailBatch-StreamThread-1]*No checkpoint found for task 1_1 state store AuditTrailBatch-store1-changelog-1 with EOS turned on.* *Reinitializing the task and restore its state from the beginning.*
> 2018-10-23 08:27:54,357 INFO  org.apache.kafka.clients.consumer.internals.Fetcher          - [Consumer clientId=AuditTrailBatch-StreamThread-1-restore-consumer, groupId=]*Resetting offset for partition AuditTrailBatch-store1-changelog-1 to offset 0.*
> 2018-10-23 08:27:54,653 INFO  org.apache.kafka.streams.processor.internals.StreamThread    - stream-thread [AuditTrailBatch-StreamThread-1]*State transition from PARTITIONS_ASSIGNED to RUNNING*
> {quote}
> From the logs above, we can get the procedure for thread AuditTrailBatch-StreamThread-1:
>  # the previous running task assigned to thread 1 is task 0_5 (the corresponding partition is AuditTrailBatch-0-5)
>  # group begins to rebalance, the new task 1_1 is assigned to thread 1.
>  # no checkpoint was found under 1_1 state folder, so reset the offset to 0 and clean the local state folder.
>  # thread 1 transitions to RUNNING state directly without the restoration for task 1_1, so the historic data for state 1_1 is lost for thread 1. 
> *ThoubleShoot*
> To investigate the cause for this issue, we analysis the source code in KafkaStream and found the key is the variable named "completedRestorers".
> This is the definition of the variable:
> {code:java}
> private final Set<TopicPartition> completedRestorers = new HashSet<>();{code}
> Each thread object has its own completedRestorers, which is created in the thread initialization, and not accessed crossly by other threads. The completedRestorers is used to record the partitions that has been restored completely in the thread.
> {code:java}
> if (restorer.hasCompleted(pos, endOffsets.get(partition))) {
>   restorer.restoreDone();
>   endOffsets.remove(partition);
>   completedRestorers.add(partition);
> }{code}
> Once the partition is added to completedRestorers set, it will be returned by restore() and pass to the next caller updateRestored(), and then the transitionToRunning() will set this task to running state. 
> But we found that completedRestorers *never* be cleared during the life cycle of this thread, even in the reset function:
> {code:java}
> //org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
> @Override
> public void reset() {
>  partitionInfo.clear();
>  stateRestorers.clear();
>  needsRestoring.clear();
>  endOffsets.clear();
>  needsInitializing.clear();
> }
> {code}
> It will cause a problem: we assume that the task 1 once assigned to thread A, so its partition has been added to completeRestores. Then it mitigated to another thread (maybe in an different instance). After several rounds of rebalancing, it transitioned to thread A again and no checkpoint was here for some reason. The right way is to clean the state folder and restore it for beginning, but now, it found this task's partition is already in completedRestorers list, so it will consider this task as restored completely and resumed running directly.
> To avoid it, we should clean the historical completedRestorers set every time after reassignment. So I add the clear operation in the reset() and validate it works.
> {code:java}
> //org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
> public void reset() {
>   partitionInfo.clear();
>   stateRestorers.clear();
>   needsRestoring.clear();
>   endOffsets.clear();
>   needsInitializing.clear();
>   //add by linyli
>   completedRestorers.clear();
>  }
> {code}
>  
> *PS:*
> In addition, I also investigate why no checkpoint was found for this state sometimes, and I found that the most common sense is when a task is mitigate from one thread to another thread in the *same* instance.
> Why?
> From source code about task reassignment, we know that the task needs write to its checkpoint file in EOS when it's closed by the previous thread, and the next thread will create the task and read from the checkpoint file for restoration. But the read/write process for this checkpoint file is Asynchronous! So it's most probably that the next thread read before the previous one finished writing, causing no checkpoint found issue and need extra restoration, which is totally a waste of time and network.
> To avoid the concurrency of read/write, I advise to add some wait time when read checkpoint to restore.
> This is my fix: 
> {code:java}
> //org/apache/kafka/streams/processor/internals/AbstractStateManager.java
> AbstractStateManager(final File baseDir,
>  final boolean eosEnabled) {
>  this.baseDir = baseDir;
>  this.eosEnabled = eosEnabled;
>  //add by linyli to fix checkpoint file latency in the same instance.
>  try
>  {
>    File checkpointfile = new File(baseDir, CHECKPOINT_FILE_NAME);
>    if(!checkpointfile.exists()) {
>      Thread.sleep(1000);
>  }
>  }catch (InterruptedException e)
>  {
>  }
>  this.checkpoint = new OffsetCheckpoint(new File(baseDir, CHECKPOINT_FILE_NAME));
> }{code}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)