[server] Handle corrupt log data after unclean shutdown to prevent TabletServer startup failure

[platinumhamburg]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

0.9.0 (latest release)

Please describe the bug 🐞

After an unclean shutdown, log files may contain incomplete batch headers. This caused LogSegment.readNextOffset() to throw NoSuchElementException via Iterables.getLast() on empty batches, crashing the TabletServer on restart.

Solution

• Add CorruptLogException to identify unrecoverable log corruption
• Fix LogSegment.readNextOffset() to guard against empty batch iterators
• Wrap LogLoader.load() to convert data-corruption-related exceptions (IOException, CorruptIndexException, LogSegmentOffsetOverflowException, InvalidOffsetException) into CorruptLogException
• Catch CorruptLogException in LogManager to delete corrupted local data and allow re-replication from leader

Are you willing to submit a PR?

• I'm willing to submit a PR!