kowshik commented on a change in pull request #11345: URL: https://github.com/apache/kafka/pull/11345#discussion_r759942807
########## File path: core/src/main/scala/kafka/log/UnifiedLog.scala ########## @@ -1498,25 +1498,25 @@ class UnifiedLog(@volatile var logStartOffset: Long, producerStateManager.takeSnapshot() updateHighWatermarkWithLogEndOffset() // Schedule an asynchronous flush of the old segment - scheduler.schedule("flush-log", () => flush(newSegment.baseOffset)) + scheduler.schedule("flush-log", () => flushUptoOffsetExclusive(newSegment.baseOffset)) newSegment } /** * Flush all local log segments * - * @param inclusive should be true during a clean shutdown, and false otherwise. The reason is that + * @param forceFlushActiveSegment should be true during a clean shutdown, and false otherwise. The reason is that * we have to pass logEngOffset + 1 to the `localLog.flush(offset: Long): Unit` function to flush empty * active segments, which is important to make sure we don't lose the empty index file during shutdown. Review comment: IIUC the main reason we would like to force flush the active segment is slightly different. Instead of `which is important to make sure we don't lose the empty index file during shutdown.`, could the comment say `which is important to make sure we persist the active segment file during shutdown, particularly when its empty.` ? ########## File path: core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala ########## @@ -1630,6 +1630,20 @@ class UnifiedLogTest { assertThrows(classOf[OffsetOutOfRangeException], () => LogTestUtils.readLog(log, 1026, 1000)) } + @Test + def testFlushingEmptyActiveSegments(): Unit = { Review comment: Sounds good. ########## File path: core/src/main/scala/kafka/log/UnifiedLog.scala ########## @@ -1498,25 +1498,25 @@ class UnifiedLog(@volatile var logStartOffset: Long, producerStateManager.takeSnapshot() updateHighWatermarkWithLogEndOffset() // Schedule an asynchronous flush of the old segment - scheduler.schedule("flush-log", () => flush(newSegment.baseOffset)) + scheduler.schedule("flush-log", () => flushUptoOffsetExclusive(newSegment.baseOffset)) newSegment } /** * Flush all local log segments * - * @param inclusive should be true during a clean shutdown, and false otherwise. The reason is that + * @param forceFlushActiveSegment should be true during a clean shutdown, and false otherwise. The reason is that * we have to pass logEngOffset + 1 to the `localLog.flush(offset: Long): Unit` function to flush empty Review comment: s/logEngOffset/logEndOffset -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org