There are two parts to this patch :
A. Move the sanity check to detect corrupt index files from OffsetIndex constructor to Log constructor below the recovery logic. In case of a hard kill, checking for corrupt index files before the last segment has been recovered will fail the require() assertion.
B. The following corner case is possible:
1. A broker rolled a new log segment file and an index file of non-zero size, and got hard killed before any appends to the index file were flushed.
2. When the broker reboots and tries to load existing log segments, it will encounter this index file that has non-zero size, but has no data.
3. Since the broker was hard killed, it will enter the recovery logic in Log.loadLogSegments().
4. The recovery logic will try to truncate the index file to the base offset of the segment. It will try to find the indexSlotFor(baseOffset). indexForSlot() will return a non- zero value, because the relativeOffset(idx, mid) == relOffset == 0.
5. This will set the size of index file to a non-zero value (which will be half of its original size which was maxIndexSize * 8).
6. Thus, the require() check for corrupted index file in Log constructor will not pass since we have #entries == size != 0 && lastOffset == baseOffset.
The solution is to modify indexSlotFor() such that it returns -1 for non–zero sized index file whose lastOffset is 0 (assuming that setLength() will set empty bytes to 0), so that the index file is truncated to #entries == size == 0.
1. Unit tests passed.
2. Change the flush interval and index append interval to really low values. Produce data using console producer (index file will have flushed entries), hard kill the broker, restart the broker. Should see the exception without A. Should pass with A, ctrl+C the broker.
3. Cleanup the kafka-logs directory, don't cleanup the zookeeper. Restart the broker (to create empty log and index files for topics created in 2 above), it will boot up, hard kill it. Restart the broker again, it should fail without B, should boot successfully with B.