Repository: apache/kafka
File: core/src/main/scala/kafka/tools/DumpLogSegments.scala (Scala)

Code before:
        println(s"Found $trailingBytes invalid bytes at the end of ${file.getName}")
    } finally fileRecords.closeHandlers()
  }

  private def printControlRecord(record: Record): Unit = {
    ControlRecordType.parse(record.key) match {
      case ControlRecordType.ABORT | ControlRecordType.COMMIT =>
        val endTxnMarker = EndTransactionMarker.deserialize(record)
        print(s" endTxnMarker: ${endTxnMarker.controlType} coordinatorEpoch: ${endTxnMarker.coordinatorEpoch}")
      case ControlRecordType.LEADER_CHANGE =>
        val leaderChangeMessage = ControlRecordUtils.deserializeLeaderChangeMessage(record)
        print(s" LeaderChange: ${LeaderChangeMessageJsonConverter.write(leaderChangeMessage, leaderChangeMessage.version())}")
      case ControlRecordType.SNAPSHOT_HEADER =>
        val header = ControlRecordUtils.deserializeSnapshotHeaderRecord(record)
        print(s" SnapshotHeader ${SnapshotHeaderRecordJsonConverter.write(header, header.version())}")
      case ControlRecordType.SNAPSHOT_FOOTER =>
        val footer = ControlRecordUtils.deserializeSnapshotFooterRecord(record)
        print(s" SnapshotFooter ${SnapshotFooterRecordJsonConverter.write(footer, footer.version())}")
      case ControlRecordType.KRAFT_VERSION =>
        val kraftVersion = ControlRecordUtils.deserializeKRaftVersionRecord(record)
        print(s" KRaftVersion ${KRaftVersionRecordJsonConverter.write(kraftVersion, kraftVersion.version())}")
      case ControlRecordType.KRAFT_VOTERS=>
        val voters = ControlRecordUtils.deserializeVotersRecord(record)
        print(s" KRaftVoters ${VotersRecordJsonConverter.write(voters, voters.version())}")
      case controlType =>
        print(s" controlType: $controlType")
    }
  }

  private def printBatchLevel(batch: FileLogInputStream.FileChannelRecordBatch, accumulativeBytes: Long): Unit = {
    if (batch.magic >= RecordBatch.MAGIC_VALUE_V2)
      print("baseOffset: " + batch.baseOffset + " lastOffset: " + batch.lastOffset + " count: " + batch.countOrNull +
        " baseSequence: " + batch.baseSequence + " lastSequence: " + batch.lastSequence +
        " producerId: " + batch.producerId + " producerEpoch: " + batch.producerEpoch +
        " partitionLeaderEpoch: " + batch.partitionLeaderEpoch + " isTransactional: " + batch.isTransactional +
        " isControl: " + batch.isControlBatch + " deleteHorizonMs: " + batch.deleteHorizonMs)
    else
      print("offset: " + batch.lastOffset)

    println(" position: " + accumulativeBytes + " " + batch.timestampType + ": " + batch.maxTimestamp +
      " size: " + batch.sizeInBytes + " magic: " + batch.magic +
      " compresscodec: " + batch.compressionType.name + " crc: " + batch.checksum + " isvalid: " + batch.isValid)
  }

  class TimeIndexDumpErrors {
    val misMatchesForTimeIndexFilesMap: mutable.Map[String, ArrayBuffer[(Long, Long)]] = mutable.Map[String, ArrayBuffer[(Long, Long)]]()
    val outOfOrderTimestamp: mutable.Map[String, ArrayBuffer[(Long, Long)]] = mutable.Map[String, ArrayBuffer[(Long, Long)]]()
    val shallowOffsetNotFound: mutable.Map[String, ArrayBuffer[(Long, Long)]] = mutable.Map[String, ArrayBuffer[(Long, Long)]]()

    def recordMismatchTimeIndex(file: File, indexTimestamp: Long, logTimestamp: Long): Unit = {
      val misMatchesSeq = misMatchesForTimeIndexFilesMap.getOrElse(file.getAbsolutePath, new ArrayBuffer[(Long, Long)]())

Review comment (bug):
Since `parseTypeId` is now private and callers use `parse()` instead, `DumpLogSegments.printControlRecord` and `MemoryRecordsBuilder.appendControlRecordWithOffset` no longer have access to the raw `typeId`. This means the numeric      
  `typeId` is no longer included in the output/exception message (e.g., `UNKNOWN(337)` becomes `UNKNOWN`). Would it be acceptable to lose this debug info, or should we keep `parseTypeId` accessible? 

Code after:
        println(s"Found $trailingBytes invalid bytes at the end of ${file.getName}")
    } finally fileRecords.closeHandlers()
  }

  private def printControlRecord(record: Record): Unit = {
    val controlTypeId = ControlRecordType.parseTypeId(record.key)
    ControlRecordType.fromTypeId(controlTypeId) match {
      case ControlRecordType.ABORT | ControlRecordType.COMMIT =>
        val endTxnMarker = EndTransactionMarker.deserialize(record)
        print(s" endTxnMarker: ${endTxnMarker.controlType} coordinatorEpoch: ${endTxnMarker.coordinatorEpoch}")
      case ControlRecordType.LEADER_CHANGE =>
        val leaderChangeMessage = ControlRecordUtils.deserializeLeaderChangeMessage(record)
        print(s" LeaderChange: ${LeaderChangeMessageJsonConverter.write(leaderChangeMessage, leaderChangeMessage.version())}")
      case ControlRecordType.SNAPSHOT_HEADER =>
        val header = ControlRecordUtils.deserializeSnapshotHeaderRecord(record)
        print(s" SnapshotHeader ${SnapshotHeaderRecordJsonConverter.write(header, header.version())}")
      case ControlRecordType.SNAPSHOT_FOOTER =>
        val footer = ControlRecordUtils.deserializeSnapshotFooterRecord(record)
        print(s" SnapshotFooter ${SnapshotFooterRecordJsonConverter.write(footer, footer.version())}")
      case ControlRecordType.KRAFT_VERSION =>
        val kraftVersion = ControlRecordUtils.deserializeKRaftVersionRecord(record)
        print(s" KRaftVersion ${KRaftVersionRecordJsonConverter.write(kraftVersion, kraftVersion.version())}")
      case ControlRecordType.KRAFT_VOTERS=>
        val voters = ControlRecordUtils.deserializeVotersRecord(record)
        print(s" KRaftVoters ${VotersRecordJsonConverter.write(voters, voters.version())}")
      case controlType =>
        print(s" controlType: $controlType($controlTypeId)")
    }
  }

  private def printBatchLevel(batch: FileLogInputStream.FileChannelRecordBatch, accumulativeBytes: Long): Unit = {
    if (batch.magic >= RecordBatch.MAGIC_VALUE_V2)
      print("baseOffset: " + batch.baseOffset + " lastOffset: " + batch.lastOffset + " count: " + batch.countOrNull +
        " baseSequence: " + batch.baseSequence + " lastSequence: " + batch.lastSequence +
        " producerId: " + batch.producerId + " producerEpoch: " + batch.producerEpoch +
        " partitionLeaderEpoch: " + batch.partitionLeaderEpoch + " isTransactional: " + batch.isTransactional +
        " isControl: " + batch.isControlBatch + " deleteHorizonMs: " + batch.deleteHorizonMs)
    else
      print("offset: " + batch.lastOffset)

    println(" position: " + accumulativeBytes + " " + batch.timestampType + ": " + batch.maxTimestamp +
      " size: " + batch.sizeInBytes + " magic: " + batch.magic +
      " compresscodec: " + batch.compressionType.name + " crc: " + batch.checksum + " isvalid: " + batch.isValid)
  }

  class TimeIndexDumpErrors {
    val misMatchesForTimeIndexFilesMap: mutable.Map[String, ArrayBuffer[(Long, Long)]] = mutable.Map[String, ArrayBuffer[(Long, Long)]]()
    val outOfOrderTimestamp: mutable.Map[String, ArrayBuffer[(Long, Long)]] = mutable.Map[String, ArrayBuffer[(Long, Long)]]()
    val shallowOffsetNotFound: mutable.Map[String, ArrayBuffer[(Long, Long)]] = mutable.Map[String, ArrayBuffer[(Long, Long)]]()

    def recordMismatchTimeIndex(file: File, indexTimestamp: Long, logTimestamp: Long): Unit = {