KAFKA-10101: Fix edge cases in Log.recoverLog and LogManager.loadLogs

[ijuma]

1. Don't advance recovery point in recoverLog unless there was a clean
   shutdown.
2. Ensure the recovery point is not ahead of the log end offset.
3. Clean and flush leader epoch cache and truncate produce state manager
   if deleting segments due to log end offset being smaller than log start
   offset.
4. If we are unable to delete clean shutdown file that exists, mark the
   directory as offline (this was the intent, but the code was wrong).

Updated one test that was failing after this change to verify the new behavior.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[ijuma]

Tests still have to be written.

[ijuma]

@junrao does this seem right? If we don't have any segments, we should be able to completely truncate the producer state manager, right?

[hachikuji]

Hmm, not sure about this. After KIP-360, we try to retain producer state as long as possible even when the corresponding entries have been removed from the log. However, we're in a strange state given that some of the later segments were apparently removed. Perhaps it is safer to treat this more like a new replica which is starting from scratch.

[ijuma]

Yeah, that's what I was thinking. What's the best way to achieve this (treat this more like a new replica which is starting from scratch)?

[rite2nikhil]

Do we still need to flush logs after recovery ?https://github.com/apache/kafka/blob/7af645b8e79ca535ce2df202c4fe8ea400a8a60e/core/src/main/scala/kafka/log/Log.scala#L804

[ijuma]

@rite2nikhil We don't need the flush the segments since we don't increment the recovery point. However, I'm not sure about the leaderEpochCache. Offline, @junrao said we didn't have to update it, but I don't fully understand why we do in the case where we delete all the log segments due to the log end offset being smaller than the log start offset.

@junrao can you please clarify?

[hachikuji]

nit: this is a big initializer. Are there parts we could move to a method?

[ijuma]

I can try.

[hachikuji]

Hmm, not sure about this. After KIP-360, we try to retain producer state as long as possible even when the corresponding entries have been removed from the log. However, we're in a strange state given that some of the later segments were apparently removed. Perhaps it is safer to treat this more like a new replica which is starting from scratch.

[hachikuji]

I guess it is because of the semantics of DeleteRecords that we trust the checkpoint over the segment data. Might be worth a comment about that since it is a bit surprising.

[ijuma]

I was also surprised, so I agree. :) Will do.

[hachikuji]

Just checking, but the issue here is that we might mistakenly mark the directory is offline if the clean shutdown file did not exist?

[ijuma]

File.delete doesn't throw an exception and we don't check the result. So the previous code was very misleading. That's what I was trying to fix. And to avoid introducing an issue if the file did not exist, I am using deleteIfExists.

[hachikuji]

Hmm, logEndOffset is defined by nextOffsetMetadata, which is initialized after loadSegments returns. But recoverLog is called within loadSegments. So does this check work as expected or am I missing something?

[ijuma]

I meant to use logEndOffsetOption, so this is a bug. :) It probably indicates that the variable name is bad (I kept it from before). If I had used the right variable, it would be:

def readNextOffset: Long = {
    val fetchData = read(offsetIndex.lastOffset, log.sizeInBytes)
    if (fetchData == null)
      baseOffset
    else
      fetchData.records.batches.asScala.lastOption
        .map(_.nextOffset)
        .getOrElse(baseOffset)
  }

The idea is that if we delete a bunch of segments, then the recovery point we passed to the Log constructor could be ahead of what remains.

[hachikuji]

Can you help me understand what was wrong with this?

[ijuma]

Jun explained in the JIRA, The concern is that if there is a hard failure during recovery, you could end up with a situation where we persisted this, but we did not flush some of the segments. Does that make sense?

[hachikuji]

Got it. So we may still be able to reopen an unflushed segment. That makes sense.

[purplefox]

Is it possible that a consumer could see "phantom" messages after recovery, even with this change?

1. Kafka Process dies with log data in page cache but not fsync'd
2. Recovery process sees the un-fsync'd log data but it looks ok so recovery succeeds, nothing to do.
3. Consumer fetches this data
4. OS hard dies, losing page cache
5. Broker is restarted and consumer tries to repeat fetch from same offset but data has gone.

It seems to me once recovery has run we should be sure that all log segments are persistently stored. I'm not sure if we're currently providing that guarantee. It would be pretty simple just to fsync each segment we recover.

[ijuma]

@purplefox To clarify, the case you are talking about is:

1. A catastrophic scenario where a partition is offline (i.e. all replicas are down)
2. The OS was not shutdown
3. The OS did not flush the data to disk (this typically happens irrespective of our flushes due to OS configurations)
4. The replica that was the last member of the ISR comes back up, registers, to ZK and the Controller makes it the
   leader (since it was the last member of the ISR, if it was a different replica, it won't be given leadership without
   unclean leader election)
5. The hw is beyond the flushed segments
6. Consumer fetches the data beyond the flushed segments
7. OS hard dies

This is an interesting edge case, it seems incredibly unlikely, but possible if the hw can be beyond the flushed segments. @junrao & @hachikuji are we missing any detail that protects us against this?

The following has a similar impact:

1. Leader accepts a write
2. Write is replicated, hw is incremented, but data is not flushed
3. All replicas die, but the ISR is not shrunk yet
4. Leader receives a write, accepts it, replication doesn't happen since replicas are gone
5. ISR is shrunk, hw is incremented
6. Producer won't receive a successful ack given min.isr=2, but the consumer reads data that is only in the leader
7. Leader crashes and the unflushed data is gone (or hard disk dies and all the data in the leader is gone)

Flushing the segments during recovery helps on some scenarios, but not the one I just mentioned (assuming I am not missing anything). @hachikuji had a "strict min isr" proposal where the ISR is never allowed to shrink below min.isr. I haven't thought about all the details, but perhaps that covers both issues. Thoughts @hachikuji?

[junrao]

For the case that Tim mentioned, if we defer advancing the recovery point, at step 5, the broker will be forced to do log recovery for all unflushed data. If the data is corrupted on disk, it will be detected during recovery.

For the other case that Ismael mentioned, it is true that data can be lost in that case, but then this is the case where all replicas have failed.

[hachikuji]

I think it is a gap that there is no minimum replication factor before a write can get exposed. Any writes that end up seeing the NOT_ENOUGH_REPLICAS_AFTER_APPEND error code are more vulnerable. These are unacknowledged writes, and the producer is expected to retry, but the consumer can still read them once the ISR shrinks and we would still view it as "data loss" if the broker failed before they could be flushed to disk. With the "strict min isr" proposal, the leader is not allowed to shrink the ISR lower than some replication factor, which helps to plug this hole.

Going back to @purplefox's suggestion, it does seem like a good idea to flush segments beyond the recovery point during recovery. It kind of serves to constrain the initial state of the system which makes it easier to reason about (e.g. you only need to worry about the loss of unflushed data from the last restart). Some of the flush weaknesses probably still exist though regardless of this change.

[ijuma]

We discussed this offline and we decided to stick with the fix in this PR for now and to file a separate JIRA to consider flushing unflushed segments during recovery. That would provide stronger guarantees after a restart.

[ijuma]

Filed https://issues.apache.org/jira/browse/KAFKA-12386.

[junrao]

@ijuma : We don't need to flush leaderEpochCache after segment recovery since new leader epochs are added through LeaderEpochFileCache.assign() which does flushing already.

[junrao]

@ijuma : Thanks for the PR. LGTM

[ijuma]

Tests passed, merged to trunk and 2.8.