Repository: apache/kafka
File: core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala (Scala)

Code before:

            val memRecords = (fetchDataInfo.records: @unchecked) match {
              case records: MemoryRecords => records
              case fileRecords: FileRecords =>
                val sizeInBytes = fileRecords.sizeInBytes
                val bytesNeeded = Math.max(config.transactionLogLoadBufferSize, sizeInBytes)

                // minOneMessage = true in the above log.read means that the buffer may need to be grown to ensure progress can be made
                if (buffer.capacity < bytesNeeded) {
                  if (config.transactionLogLoadBufferSize < bytesNeeded)
                    warn(s"Loaded transaction metadata from $topicPartition with buffer larger ($bytesNeeded bytes) than " +
                      s"configured transaction.state.log.load.buffer.size (${config.transactionLogLoadBufferSize} bytes)")

                  buffer = ByteBuffer.allocate(bytesNeeded)
                } else {
                  buffer.clear()
                }
                buffer.clear()
                fileRecords.readInto(buffer, 0)
                MemoryRecords.readableRecords(buffer)
            }
            memRecords.batches.forEach { batch =>
              for (record <- batch.asScala) {
                require(record.hasKey, "Transaction state log's key should not be null")
                try {
                  val transactionalId = TransactionLog.readTxnRecordKey(record.key).asInstanceOf[String]
                  // load transaction metadata along with transaction state
                  val txnMetadata = TransactionLog.readTxnRecordValue(transactionalId, record.value)
                  if (txnMetadata == null) {
                    loadedTransactions.remove(transactionalId)
                  } else {
                    loadedTransactions.put(transactionalId, txnMetadata)
                  }
                } catch {
                  case e: IllegalStateException =>
                    warn(s"Unknown message key version while loading transaction state from $topicPartition. " +
                      s"Ignoring it. It could be a left over from an aborted upgrade. Error: ${e.getMessage}")
                }
              }
              currOffset = batch.nextOffset
            }
          }
        } catch {
          case t: Throwable => error(s"Error loading transactions from transaction log $topicPartition", t)
        }
    }

    loadedTransactions
  }

  /**

Review comment (suggestion):
```scala
                val transactionalId = try Some(TransactionLog.readTxnRecordKey(record.key))
                catch {
                  case e: IllegalStateException =>
                    warn(s"Unknown message key version while loading transaction state from $topicPartition. " +
                      s"Ignoring it. It could be a left over from an aborted upgrade. Error: ${e.getMessage}")
                    None
                }
                transactionalId.foreach { txnId =>
                  // load transaction metadata along with transaction state
                  val txnMetadata = TransactionLog.readTxnRecordValue(txnId, record.value)
                  if (txnMetadata == null) {
                    loadedTransactions.remove(txnId)
                  } else {
                    loadedTransactions.put(txnId, txnMetadata)
                  }
                }
```

Code after:

            val memRecords = (fetchDataInfo.records: @unchecked) match {
              case records: MemoryRecords => records
              case fileRecords: FileRecords =>
                val sizeInBytes = fileRecords.sizeInBytes
                val bytesNeeded = Math.max(config.transactionLogLoadBufferSize, sizeInBytes)

                // minOneMessage = true in the above log.read means that the buffer may need to be grown to ensure progress can be made
                if (buffer.capacity < bytesNeeded) {
                  if (config.transactionLogLoadBufferSize < bytesNeeded)
                    warn(s"Loaded transaction metadata from $topicPartition with buffer larger ($bytesNeeded bytes) than " +
                      s"configured transaction.state.log.load.buffer.size (${config.transactionLogLoadBufferSize} bytes)")

                  buffer = ByteBuffer.allocate(bytesNeeded)
                } else {
                  buffer.clear()
                }
                buffer.clear()
                fileRecords.readInto(buffer, 0)
                MemoryRecords.readableRecords(buffer)
            }
            memRecords.batches.forEach { batch =>
              for (record <- batch.asScala) {
                require(record.hasKey, "Transaction state log's key should not be null")
                val transactionalId = try Some(TransactionLog.readTxnRecordKey(record.key))
                catch {
                  case e: IllegalStateException =>
                    warn(s"Unknown message key version while loading transaction state from $topicPartition. " +
                      s"Ignoring it. It could be a left over from an aborted upgrade", e)
                    None
                }
                transactionalId.foreach { txnId =>
                  // load transaction metadata along with transaction state
                  val txnMetadata = TransactionLog.readTxnRecordValue(txnId, record.value)
                  if (txnMetadata == null) {
                    loadedTransactions.remove(txnId)
                  } else {
                    loadedTransactions.put(txnId, txnMetadata)
                  }
                }
              }
              currOffset = batch.nextOffset
            }
          }
        } catch {
          case t: Throwable => error(s"Error loading transactions from transaction log $topicPartition", t)
        }
    }

    loadedTransactions
  }