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

Code before:
    logProps.put(TopicConfig.RETENTION_MS_CONFIG, log.config.retentionMs: java.lang.Long)
    logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT)
    logProps.put(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG, 0: Integer)
    val config = new LogConfig(logProps)
    log.updateConfig(config)

    // log cleanup inprogress, the log is not available for compaction
    val cleanable = cleanerManager.grabFilthiestCompactedLog(time, new PreCleanStats()).toScala
    assertEquals(0, cleanable.size, "should have 0 logs ready to be compacted")

    // log cleanup finished, and log can be picked up for compaction
    cleanerManager.resumeCleaning(deletableLog.asScala.map(_.getKey).toSet.asJava)
    val cleanable2 = cleanerManager.grabFilthiestCompactedLog(time, new PreCleanStats()).toScala
    assertEquals(1, cleanable2.size, "should have 1 logs ready to be compacted")

    // update cleanup policy to delete
    logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE)
    val config2 = new LogConfig(logProps)
    log.updateConfig(config2)

    // compaction in progress, should have 0 log eligible for log cleanup
    val deletableLog2 = cleanerManager.pauseCleaningForNonCompactedPartitions()
    assertEquals(0, deletableLog2.size, "should have 0 logs ready to be deleted")

    // compaction done, should have 1 log eligible for log cleanup
    cleanerManager.doneDeleting(util.List.of(cleanable2.get.topicPartition))
    val deletableLog3 = cleanerManager.pauseCleaningForNonCompactedPartitions()
    assertEquals(1, deletableLog3.size, "should have 1 logs ready to be deleted")
  }

  @Test
  def testUpdateCheckpointsShouldAddOffsetToPartition(): Unit = {
    val records = TestUtils.singletonRecords("test".getBytes, key="test".getBytes)
    val log: UnifiedLog = createLog(records.sizeInBytes * 5, TopicConfig.CLEANUP_POLICY_COMPACT)
    val cleanerManager: LogCleanerManager = createCleanerManager(log)

    // expect the checkpoint offset is not the expectedOffset before doing updateCheckpoints
    assertNotEquals(offset, cleanerManager.allCleanerCheckpoints.getOrDefault(topicPartition, 0))

    cleanerManager.updateCheckpoints(logDir, Optional.of(util.Map.entry(topicPartition, offset)), Optional.empty())
    // expect the checkpoint offset is now updated to the expected offset after doing updateCheckpoints
    assertEquals(offset, cleanerManager.allCleanerCheckpoints.get(topicPartition))
  }

  @Test
  def testUpdateCheckpointsShouldRemovePartitionData(): Unit = {
    val records = TestUtils.singletonRecords("test".getBytes, key="test".getBytes)
    val log: UnifiedLog = createLog(records.sizeInBytes * 5, TopicConfig.CLEANUP_POLICY_COMPACT)
    val cleanerManager: LogCleanerManager = createCleanerManager(log)

    // write some data into the cleaner-offset-checkpoint file

Review comment (suggestion):
Could we change `cleanerCheckpoints` to `java.util.HashMap` ? (L63)

Code after:
    logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT)
    logProps.put(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG, 0: Integer)
    val config = new LogConfig(logProps)
    log.updateConfig(config)

    // log cleanup inprogress, the log is not available for compaction
    val cleanable = cleanerManager.grabFilthiestCompactedLog(time, new PreCleanStats()).toScala
    assertEquals(0, cleanable.size, "should have 0 logs ready to be compacted")

    // log cleanup finished, and log can be picked up for compaction
    cleanerManager.resumeCleaning(
      deletableLog.stream()
        .map[TopicPartition](entry => entry.getKey)
        .collect(Collectors.toSet[TopicPartition]())
    )
    val cleanable2 = cleanerManager.grabFilthiestCompactedLog(time, new PreCleanStats()).toScala
    assertEquals(1, cleanable2.size, "should have 1 logs ready to be compacted")

    // update cleanup policy to delete
    logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE)
    val config2 = new LogConfig(logProps)
    log.updateConfig(config2)

    // compaction in progress, should have 0 log eligible for log cleanup
    val deletableLog2 = cleanerManager.pauseCleaningForNonCompactedPartitions()
    assertEquals(0, deletableLog2.size, "should have 0 logs ready to be deleted")

    // compaction done, should have 1 log eligible for log cleanup
    cleanerManager.doneDeleting(util.List.of(cleanable2.get.topicPartition))
    val deletableLog3 = cleanerManager.pauseCleaningForNonCompactedPartitions()
    assertEquals(1, deletableLog3.size, "should have 1 logs ready to be deleted")
  }

  @Test
  def testUpdateCheckpointsShouldAddOffsetToPartition(): Unit = {
    val records = TestUtils.singletonRecords("test".getBytes, key="test".getBytes)
    val log: UnifiedLog = createLog(records.sizeInBytes * 5, TopicConfig.CLEANUP_POLICY_COMPACT)
    val cleanerManager: LogCleanerManager = createCleanerManager(log)

    // expect the checkpoint offset is not the expectedOffset before doing updateCheckpoints
    assertNotEquals(offset, cleanerManager.allCleanerCheckpoints.getOrDefault(topicPartition, 0))

    cleanerManager.updateCheckpoints(logDir, Optional.of(util.Map.entry(topicPartition, offset)), Optional.empty())
    // expect the checkpoint offset is now updated to the expected offset after doing updateCheckpoints
    assertEquals(offset, cleanerManager.allCleanerCheckpoints.get(topicPartition))
  }

  @Test
  def testUpdateCheckpointsShouldRemovePartitionData(): Unit = {
    val records = TestUtils.singletonRecords("test".getBytes, key="test".getBytes)
    val log: UnifiedLog = createLog(records.sizeInBytes * 5, TopicConfig.CLEANUP_POLICY_COMPACT)