[KAFKA-10432] LeaderEpochCache is incorrectly recovered on segment recovery for epoch 0

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 2.3.0, 2.4.0, 2.5.0, 2.6.0
• Fix Version/s: 2.6.1
• Assignee: Lucas Bradstreet
• Reporter: Lucas Bradstreet
• Created: 25/Aug/20 17:01
• Updated: 20/Dec/22 00:25
• Resolved: 08/Sep/20 19:50

Description

I added some functionality to the system tests to compare epoch cache lineages (https://github.com/apache/kafka/pull/9213), and I found a bug in leader epoch cache recovery.

The test hard kills a broker and the cache hasn't been flushed yet, and then it starts up and goes through log recovery. After recovery there is divergence in the epoch caches for epoch 0:

AssertionError: leader epochs for output-topic-1 didn't match
 [{0: 9393L, 2: 9441L, 4: 42656L},
 {0: 0L, 2: 9441L, 4: 42656L}, 
 {0: 0L, 2: 9441L, 4: 42656L}]                                                                                                                                                                

The cache is supposed to include the offset for epoch 0 but in recovery it skips it https://github.com/apache/kafka/blob/487b3682ebe0eefde3445b37ee72956451a9d15e/core/src/main/scala/kafka/log/LogSegment.scala#L364 due to https://github.com/apache/kafka/commit/d152989f26f51b9004b881397db818ad6eaf0392. Then it stamps the epoch with a later offset when fetching from the leader.

I'm not sure why the recovery code includes the condition `batch.partitionLeaderEpoch > 0`. I discussed this with Jason Gustafson and he believes it may have been intended to avoid assigning negative epochs but is not sure why it was added. None of the tests fail with this check removed.

          leaderEpochCache.foreach { cache =>
            if (batch.partitionLeaderEpoch > 0 && cache.latestEpoch.forall(batch.partitionLeaderEpoch > _))
              cache.assign(batch.partitionLeaderEpoch, batch.baseOffset)
          }

Issue Links

• links to: GitHub Pull Request #9219