KAFKA-20979: Ensure indexes resize are fsynced when shutting down

[mimaison]

This avoids reading invalid length for the index files after a clean
shut down

This does not prevent the issue happening on an unclean shutdown. For
validating indexes at startup, we have
https://issues.apache.org/jira/browse/KAFKA-19200

Reviewers: Gaurav Narula gaurav_narula2@apple.com, Luke Chen
showuon@gmail.com

[gaurav-narula]

Looking at the javadoc for FileChannel#force

This method is only guaranteed to force changes that were made to this channel's file via the methods defined in this class, or the methods defined by java.io.FileOutputStream or java.io.RandomAccessFile when the channel was obtained with the getChannel method. It may or may not force changes that were made by modifying the content of a mapped byte buffer obtained by invoking the map method. Invoking the force method of the mapped byte buffer will force changes made to the buffer's content to be written.

I think if the intention is to sync both the existing data and the file metadata we should invoke mmap.force() just before safeForceUnmap() in addition to this

[showuon]

Nice catch @gaurav-narula ! Yes, the FileChannel#map change won't be guaranteed to be written into the disk. I think we can directly use the existing flush() here.

[mimaison]

Maybe my comment fsync the file after resizing to ensure both content and size are durable is confusing. This method only touches metadata (length) and my concern was to ensure metadata are flushed.

The content should already have been flushed from LogSegment.flush().

[gaurav-narula]

Yes, it was that comment that got me looking :)

The content should already have been flushed from LogSegment.flush()

IIUC, the flush is invoked in a thread spawned during shutdown at LogManager at

kafka/storage/src/main/java/org/apache/kafka/storage/internals/log/LogManager.java

Line 884 in 1fa9340

log.flush(true);

followed by a UnifiedLog#close.

There is however a conditional append to a time index that may happen after LogSegment#flush() and before TimeIndex#close() at LogSegment#close()

kafka/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java

Line 770 in 1fa9340

Utils.swallow(LOGGER, Level.WARN, "maybeAppend", () -> timeIndex().maybeAppend(maxTimestampSoFar(), shallowOffsetOfMaxTimestampSoFar(), true));

which is not fsynced.

[mimaison]

Right, the content could also not be flushed. I've added a call to flush() in AbstractIndex.close().

[showuon]

Thanks for the PR. I think we also need to add a test for it. We already have a AbstractIndexTest#testResizeInvokeUnmap test to verify the Unmap is invoked when resize. We can add the similar test to verify the flush method is invoked.

[showuon]

Nice catch @gaurav-narula ! Yes, the FileChannel#map change won't be guaranteed to be written into the disk. I think we can directly use the existing flush() here.

[gaurav-narula]

Thanks for the update! A small comment around NPE due to which we've failing tests

[gaurav-narula]

I think the tests are failing due to an NPE here as close() is invoked multiple times. Perhaps consider guarding this with a null check?

[mimaison]

Yes, I pushed a fix, thanks