Repository: apache/kafka
File: core/src/main/scala/kafka/raft/KafkaMetadataLog.scala (Scala)

Code before:
      defaultLogConfig,
      0L,
      0L,
      scheduler,
      new BrokerTopicStats,
      time,
      Integer.MAX_VALUE,
      new ProducerStateManagerConfig(Integer.MAX_VALUE, false),
      Integer.MAX_VALUE,
      new LogDirFailureChannel(5),
      false,
      Optional.of(topicId)
    )

    val metadataLog = new KafkaMetadataLog(
      log,
      time,
      scheduler,
      recoverSnapshots(log),
      topicPartition,
      config,
      nodeId
    )

    if (defaultLogConfig.segmentSize() < config.logSegmentBytes()) {
      metadataLog.error(s"Overriding ${MetadataLogConfig.METADATA_LOG_SEGMENT_BYTES_CONFIG} is only supported for testing. Setting " +
        s"this value too low may lead to an inability to write batches of metadata records.")
    }

    // When recovering, truncate fully if the latest snapshot is after the log end offset. This can happen to a follower
    // when the follower crashes after downloading a snapshot from the leader but before it could truncate the log fully.
    metadataLog.truncateToLatestSnapshot()

    metadataLog
  }

  private def recoverSnapshots(
    log: UnifiedLog
  ): mutable.TreeMap[OffsetAndEpoch, Option[FileRawSnapshotReader]] = {
    val snapshotsToRetain = mutable.TreeMap.empty[OffsetAndEpoch, Option[FileRawSnapshotReader]]
    val snapshotsToDelete = mutable.Buffer.empty[SnapshotPath]

    // Scan the log directory; deleting partial snapshots and older snapshot, only remembering immutable snapshots start
    // from logStartOffset
    val filesInDir = Files.newDirectoryStream(log.dir.toPath)

    try {
      filesInDir.forEach { path =>
        Snapshots.parse(path).ifPresent { snapshotPath =>
          // Collect partial snapshot, deleted snapshot and older snapshot for deletion
          if (snapshotPath.partial

Review comment (suggestion):
it should be `INTERNAL_METADATA_LOG_SEGMENT_BYTES_CONFIG`

Code after:
      defaultLogConfig,
      0L,
      0L,
      scheduler,
      new BrokerTopicStats,
      time,
      Integer.MAX_VALUE,
      new ProducerStateManagerConfig(Integer.MAX_VALUE, false),
      Integer.MAX_VALUE,
      new LogDirFailureChannel(5),
      false,
      Optional.of(topicId)
    )

    val metadataLog = new KafkaMetadataLog(
      log,
      time,
      scheduler,
      recoverSnapshots(log),
      topicPartition,
      config,
      nodeId
    )

    if (defaultLogConfig.segmentSize() < config.logSegmentBytes()) {
      metadataLog.error(s"Overriding ${MetadataLogConfig.INTERNAL_METADATA_LOG_SEGMENT_BYTES_CONFIG} is only supported for testing. Setting " +
        s"this value too low may lead to an inability to write batches of metadata records.")
    }

    // When recovering, truncate fully if the latest snapshot is after the log end offset. This can happen to a follower
    // when the follower crashes after downloading a snapshot from the leader but before it could truncate the log fully.
    metadataLog.truncateToLatestSnapshot()

    metadataLog
  }

  private def recoverSnapshots(
    log: UnifiedLog
  ): mutable.TreeMap[OffsetAndEpoch, Option[FileRawSnapshotReader]] = {
    val snapshotsToRetain = mutable.TreeMap.empty[OffsetAndEpoch, Option[FileRawSnapshotReader]]
    val snapshotsToDelete = mutable.Buffer.empty[SnapshotPath]

    // Scan the log directory; deleting partial snapshots and older snapshot, only remembering immutable snapshots start
    // from logStartOffset
    val filesInDir = Files.newDirectoryStream(log.dir.toPath)

    try {
      filesInDir.forEach { path =>
        Snapshots.parse(path).ifPresent { snapshotPath =>
          // Collect partial snapshot, deleted snapshot and older snapshot for deletion
          if (snapshotPath.partial