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

Code before:
    cleanerManager.alterCheckpointDir(topicPartition, logDir, logDir2)
    // verify we still can get the partition offset after alterCheckpointDir
    // This data should locate in logDir2, not logDir
    assertEquals(offset, cleanerManager.allCleanerCheckpoints.get(topicPartition))

    // force delete the logDir2 from checkpoints, so that the partition data should also be deleted
    cleanerManager.handleLogDirFailure(logDir2.getAbsolutePath)
    assertFalse(cleanerManager.allCleanerCheckpoints.containsKey(topicPartition))
  }

  /**
    * log under cleanup should still be eligible for log truncation
    */
  @Test
  def testConcurrentLogCleanupAndLogTruncation(): Unit = {
    val records = TestUtils.singletonRecords("test".getBytes, key="test".getBytes)
    val log: UnifiedLog = createLog(records.sizeInBytes * 5, TopicConfig.CLEANUP_POLICY_DELETE)
    val cleanerManager: LogCleanerManager = createCleanerManager(log)

    // log cleanup starts
    val pausedPartitions = cleanerManager.pauseCleaningForNonCompactedPartitions()
    // Log truncation happens due to unclean leader election
    cleanerManager.abortAndPauseCleaning(log.topicPartition)
    cleanerManager.resumeCleaning(util.Set.of(log.topicPartition))
    // log cleanup finishes and pausedPartitions are resumed
    cleanerManager.resumeCleaning(pausedPartitions.asScala.map(_.getKey).toSet.asJava)

    assertEquals(Optional.empty(), cleanerManager.cleaningState(log.topicPartition))
  }

  /**
    * log under cleanup should still be eligible for topic deletion
    */
  @Test
  def testConcurrentLogCleanupAndTopicDeletion(): Unit = {
    val records = TestUtils.singletonRecords("test".getBytes, key = "test".getBytes)
    val log: UnifiedLog = createLog(records.sizeInBytes * 5, TopicConfig.CLEANUP_POLICY_DELETE)
    val cleanerManager: LogCleanerManager = createCleanerManager(log)

    // log cleanup starts
    val pausedPartitions = cleanerManager.pauseCleaningForNonCompactedPartitions()
    // Broker processes StopReplicaRequest with delete=true
    cleanerManager.abortCleaning(log.topicPartition)
    // log cleanup finishes and pausedPartitions are resumed
    cleanerManager.resumeCleaning(pausedPartitions.asScala.map(_.getKey).toSet.asJava)

    assertEquals(Optional.empty(), cleanerManager.cleaningState(log.topicPartition))
  }

  /**
    * When looking for logs with segments ready to be deleted we shouldn't consider

Review comment (suggestion):
Could we eliminate the conversion?
Additionally, there are some similar patterns in this test; please clean them up.

Code after:
    cleanerManager.updateCheckpoints(logDir, Optional.of(util.Map.entry(topicPartition, offset)), Optional.empty())
    assertEquals(offset, cleanerManager.allCleanerCheckpoints.get(topicPartition))

    cleanerManager.alterCheckpointDir(topicPartition, logDir, logDir2)
    // verify we still can get the partition offset after alterCheckpointDir
    // This data should locate in logDir2, not logDir
    assertEquals(offset, cleanerManager.allCleanerCheckpoints.get(topicPartition))

    // force delete the logDir2 from checkpoints, so that the partition data should also be deleted
    cleanerManager.handleLogDirFailure(logDir2.getAbsolutePath)
    assertFalse(cleanerManager.allCleanerCheckpoints.containsKey(topicPartition))
  }

  /**
    * log under cleanup should still be eligible for log truncation
    */
  @Test
  def testConcurrentLogCleanupAndLogTruncation(): Unit = {
    val records = TestUtils.singletonRecords("test".getBytes, key="test".getBytes)
    val log: UnifiedLog = createLog(records.sizeInBytes * 5, TopicConfig.CLEANUP_POLICY_DELETE)
    val cleanerManager: LogCleanerManager = createCleanerManager(log)

    // log cleanup starts
    val pausedPartitions = cleanerManager.pauseCleaningForNonCompactedPartitions()
    // Log truncation happens due to unclean leader election
    cleanerManager.abortAndPauseCleaning(log.topicPartition)
    cleanerManager.resumeCleaning(util.Set.of(log.topicPartition))
    // log cleanup finishes and pausedPartitions are resumed
    cleanerManager.resumeCleaning(
      pausedPartitions.stream()
        .map[TopicPartition](entry => entry.getKey)
        .collect(Collectors.toSet[TopicPartition]())
    )

    assertEquals(Optional.empty(), cleanerManager.cleaningState(log.topicPartition))
  }

  /**
    * log under cleanup should still be eligible for topic deletion
    */
  @Test
  def testConcurrentLogCleanupAndTopicDeletion(): Unit = {
    val records = TestUtils.singletonRecords("test".getBytes, key = "test".getBytes)
    val log: UnifiedLog = createLog(records.sizeInBytes * 5, TopicConfig.CLEANUP_POLICY_DELETE)
    val cleanerManager: LogCleanerManager = createCleanerManager(log)

    // log cleanup starts
    val pausedPartitions = cleanerManager.pauseCleaningForNonCompactedPartitions()
    // Broker processes StopReplicaRequest with delete=true
    cleanerManager.abortCleaning(log.topicPartition)
    // log cleanup finishes and pausedPartitions are resumed