KAFKA-13603: Allow the empty active segment to have missing offset index during recovery

core/src/main/scala/kafka/log/UnifiedLog.scala

@@ -655,6 +655,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
   def close(): Unit = {
     debug("Closing log")
     lock synchronized {
+      flush(logEndOffset + 1)
       maybeFlushMetadataFile()
       localLog.checkIfMemoryMappedBufferClosed()
       producerExpireCheck.cancel(true)
@@ -1504,6 +1505,8 @@ class UnifiedLog(@volatile var logStartOffset: Long,
 
   /**
    * Flush all local log segments
+   * We have to pass logEngOffset + 1 to the `def flush(offset: Long): Unit` function to flush empty
+   * active segments, which is important to make sure we don't lose logEndOffset during shutdown.
    */
   def flush(): Unit = flush(logEndOffset)
 

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 = {
+    /* create a multipart log with 100 messages */
+    val logConfig = LogTestUtils.createLogConfig(segmentBytes = 100)
+    val log = createLog(logDir, logConfig)
+    val numMessages = 2
+    val messageSets = (0 until numMessages).map(i => TestUtils.singletonRecords(value = i.toString.getBytes,
+      timestamp = mockTime.milliseconds))
+    messageSets.foreach(log.appendAsLeader(_, leaderEpoch = 0))
+    log.roll()
+    log.close()
+    assertEquals(numMessages + 1, logDir.listFiles(_.getName.endsWith(".index")).length)
+  }
+
   /**
    * Test that covers reads and writes on a multisegment log. This test appends a bunch of messages
    * and then reads them all back and checks that the message read and offset matches what was appended.