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

Code before:
              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. 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)
                  }
                }
              }
              currOffset = batch.nextOffset
            }
          }
        } catch {
          case t: Throwable => error(s"Error loading transactions from transaction log $topicPartition", t)
        }
    }

    loadedTransactions
  }

  /**
   * Add a transaction topic partition into the cache

Review comment (suggestion):
```scala
                    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)
```

Code after:
              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
  }

  /**
   * Add a transaction topic partition into the cache