Repository: apache/kafka
File: core/src/test/scala/unit/kafka/log/LogManagerTest.scala (Scala)

Code before:
    val currentLog = logManager.getOrCreateLog(tp, topicId = Optional.empty)
    // Create a future log in a different directory
    logManager.maybeUpdatePreferredLogDir(tp, logDir2.getAbsolutePath)
    logManager.getOrCreateLog(tp, isFuture = true, topicId = Optional.empty)

    // Spy on the source log to verify close() is not called
    val spyCurrentLog = spy(currentLog)
    // Inject the spy into the map
    val field = classOf[LogManager].getDeclaredField("currentLogs")
    field.setAccessible(true)
    val currentLogs = field.get(logManager).asInstanceOf[ConcurrentHashMap[TopicPartition, UnifiedLog]]
    currentLogs.put(tp, spyCurrentLog)

    logManager.replaceCurrentWithFutureLog(tp)

    // Verify close() was NOT called on the source log
    verify(spyCurrentLog, never()).close()

    // Verify the source log was renamed to .delete
    assertTrue(spyCurrentLog.dir.getName.endsWith(LogFileUtils.DELETE_DIR_SUFFIX))

    // Verify that flush() can be called without error (no ClosedChannelException)
    val flushLog: Executable = () => spyCurrentLog.flush(false)
    assertDoesNotThrow(flushLog)
  }

  @Test
  def testCheckpointLogStartOffsetForRemoteTopic(): Unit = {
    logManager.shutdown()

    val props = new Properties()
    props.putAll(logProps)
    props.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true")
    val logConfig = new LogConfig(props)
    logManager = TestUtils.createLogManager(
      defaultConfig = logConfig,
      configRepository = new MockConfigRepository,
      logDirs = Seq(this.logDir),
      time = this.time,
      recoveryThreadsPerDataDir = 1,
      remoteStorageSystemEnable = true
    )

    val checkpointFile = new File(logDir, JLogManager.LOG_START_OFFSET_CHECKPOINT_FILE)
    val checkpoint = new OffsetCheckpointFile(checkpointFile, null)
    val topicPartition = new TopicPartition("test", 0)
    val log = logManager.getOrCreateLog(topicPartition, topicId = Optional.empty)
    var offset = 0L
    for(_ <- 0 until 50) {
      val set = TestUtils.singletonRecords("test".getBytes())
      val info = log.appendAsLeader(set, 0)

Review comment (suggestion):
To trigger the flush, we have to set `flushOffset > localLog.recoveryPoint()` ([here](https://github.com/apache/kafka/blob/trunk/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java#L2232-L2235)). Because there's no any data, the flushOffset is 0 and recoveryPoint is 0, too. I just tested it, and this flush will not throw exception even if we close the srcLog as before. We have to make the `flushOffset > 0` to trigger the exception, something like this:

```
   // Verify that flush() can be called without error (no ClosedChannelException)
    when(spyCurrentLog.logEndOffset()).thenReturn(100L)
    val flushLog: Executable = () => spyCurrentLog.flush(false)
    assertDoesNotThrow(flushLog)
```

Code after:
    val currentLog = logManager.getOrCreateLog(tp, topicId = Optional.empty)
    // Create a future log in a different directory
    logManager.maybeUpdatePreferredLogDir(tp, logDir2.getAbsolutePath)
    logManager.getOrCreateLog(tp, isFuture = true, topicId = Optional.empty)

    // Spy on the source log to verify close() is not called
    val spyCurrentLog = spy(currentLog)
    // Inject the spy into the map
    val field = classOf[LogManager].getDeclaredField("currentLogs")
    field.setAccessible(true)
    val currentLogs = field.get(logManager).asInstanceOf[ConcurrentHashMap[TopicPartition, UnifiedLog]]
    currentLogs.put(tp, spyCurrentLog)

    logManager.replaceCurrentWithFutureLog(tp)

    // Verify close() was NOT called on the source log
    verify(spyCurrentLog, never()).close()

    // Verify the source log was renamed to .delete
    assertTrue(spyCurrentLog.dir.getName.endsWith(LogFileUtils.DELETE_DIR_SUFFIX))

    // Verify that flush() can be called without error (no ClosedChannelException)
    // Mock logEndOffset > 0 to trigger actual flush (flush only happens when flushOffset > recoveryPoint)
    when(spyCurrentLog.logEndOffset()).thenReturn(100L)
    val flushLog: Executable = () => spyCurrentLog.flush(false)
    assertDoesNotThrow(flushLog)
  }

  @Test
  def testCheckpointLogStartOffsetForRemoteTopic(): Unit = {
    logManager.shutdown()

    val props = new Properties()
    props.putAll(logProps)
    props.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true")
    val logConfig = new LogConfig(props)
    logManager = TestUtils.createLogManager(
      defaultConfig = logConfig,
      configRepository = new MockConfigRepository,
      logDirs = Seq(this.logDir),
      time = this.time,
      recoveryThreadsPerDataDir = 1,
      remoteStorageSystemEnable = true
    )

    val checkpointFile = new File(logDir, JLogManager.LOG_START_OFFSET_CHECKPOINT_FILE)
    val checkpoint = new OffsetCheckpointFile(checkpointFile, null)
    val topicPartition = new TopicPartition("test", 0)
    val log = logManager.getOrCreateLog(topicPartition, topicId = Optional.empty)
    var offset = 0L
    for(_ <- 0 until 50) {