MINOR: Ensure LocalLog.flush() is immune to recoveryPoint change by different thread

[kowshik]

Issue:
Imagine a scenario where two threads T1 and T2 are inside UnifiedLog.flush() concurrently:

• KafkaScheduler thread T1 -> The periodic work calls LogManager.flushDirtyLogs() which in turn calls UnifiedLog.flush(). For example, this can happen due to log.flush.scheduler.interval.ms here.
• KafkaScheduler thread T2 -> A UnifiedLog.flush() call is triggered asynchronously during segment roll here.

Supposing if thread T1 advances the recovery point beyond the flush offset of thread T2, then this could trip the check within LogSegments.values() here for thread T2, when it is called from LocalLog.flush() here. The exception causes the KafkaScheduler thread to die, which is not desirable.

Fix:
We fix this by ensuring that LocalLog.flush() is immune to the case where the recoveryPoint advances beyond the flush offset.

Tests:
I was able to test this manually by introducing barriers in the code to help simulate the race condition. As such, this is a hard case to write an automated unit test for, so I haven't added a new test case in this PR. So I'm mostly just relying on code review and also ensure there are no regressions in existing tests.

[kowshik]

cc @junrao @lbradstreet for review

[junrao]

@kowshik : Thanks for the PR. LGTM. Are the test failures related?

This issue could occur even with flushing on rolled segments. In general, we could have a pool of background threads for flushing rolled log segments. It's possible for the same partition's log to be rolled quickly and flushed by different threads in parallel.

[kowshik]

@junrao Thanks for the review! I checked the test failures, and they look unrelated to this PR. I agree, your suggestion is a good way to simplify the code and it will be a lot more maintainable too. I have opened KAFKA-13701 to track the improvement.