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

Code before:
                        producerStateManagerConfig: ProducerStateManagerConfig = producerStateManagerConfig,
                        producerIdExpirationCheckIntervalMs: Int = TransactionLogConfig.PRODUCER_ID_EXPIRATION_CHECK_INTERVAL_MS_DEFAULT,
                        lastShutdownClean: Boolean = true,
                        topicId: Option[Uuid] = None,
                        keepPartitionMetadataFile: Boolean = true,
                        remoteStorageSystemEnable: Boolean = false,
                        remoteLogManager: Option[RemoteLogManager] = None,
                        logOffsetsListener: LogOffsetsListener = LogOffsetsListener.NO_OP_OFFSETS_LISTENER): UnifiedLog = {
    val log = LogTestUtils.createLog(dir, config, brokerTopicStats, scheduler, time, logStartOffset, recoveryPoint,
      maxTransactionTimeoutMs, producerStateManagerConfig, producerIdExpirationCheckIntervalMs,
      lastShutdownClean, topicId, keepPartitionMetadataFile, new ConcurrentHashMap[String, Integer],
      remoteStorageSystemEnable, remoteLogManager, logOffsetsListener)
    logsToClose = logsToClose :+ log
    log
  }

  private def createLogWithOffsetOverflow(logConfig: LogConfig): (UnifiedLog, LogSegment) = {
    LogTestUtils.initializeLogDirWithOverflowedSegment(logDir)

    val log = createLog(logDir, logConfig, recoveryPoint = Long.MaxValue)
    val segmentWithOverflow = LogTestUtils.firstOverflowSegment(log).getOrElse {
      throw new AssertionError("Failed to create log with a segment which has overflowed offsets")
    }

    (log, segmentWithOverflow)

  @Test
  def testStaleProducerEpochReturnsRecoverableErrorForTV1Clients(): Unit = {
    // Producer epoch gets incremented (coordinator fail over, completed transaction, etc.)
    // and client has stale cached epoch. Fix prevents fatal InvalidTxnStateException.

    val producerStateManagerConfig = new ProducerStateManagerConfig(86400000, true)
    val logConfig = LogTestUtils.createLogConfig(segmentBytes = 2048 * 5)
    val log = createLog(logDir, logConfig, producerStateManagerConfig = producerStateManagerConfig)

    val producerId = 123L
    val oldEpoch = 5.toShort
    val newEpoch = 6.toShort

    // Step 1: Simulate a scenario where producer epoch was incremented to fence the producer
    val previousRecords = MemoryRecords.withTransactionalRecords(
      Compression.NONE, producerId, newEpoch, 0,
      new SimpleRecord("previous-key".getBytes, "previous-value".getBytes)
    )
    val previousGuard = log.maybeStartTransactionVerification(producerId, 0, newEpoch, false)  // TV1 = supportsEpochBump = false
    log.appendAsLeader(previousRecords, 0, AppendOrigin.CLIENT, RequestLocal.noCaching, previousGuard)

    // Complete the transaction normally (commits do update producer state with current epoch)
    val commitMarker = MemoryRecords.withEndTransactionMarker(
      producerId, newEpoch, new EndTransactionMarker(ControlRecordType.COMMIT, 0)
    )

Review comment (bug):
This causes a build failure.
```
> Task :core:compileTestScala
[Error] /home/chia7712/project/kafka/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala:4739:3: Missing closing brace `}` assumed here
```

Code after:
    val producerStateManagerConfig = new ProducerStateManagerConfig(86400000, true)
    val logConfig = LogTestUtils.createLogConfig(segmentBytes = 2048 * 5)
    val log = createLog(logDir, logConfig, producerStateManagerConfig = producerStateManagerConfig)

    val producerId = 456L
    val originalEpoch = 3.toShort
    val bumpedEpoch = 4.toShort

    // Step 1: Start transaction with epoch 3 (before timeout)
    val initialRecords = MemoryRecords.withTransactionalRecords(
      Compression.NONE, producerId, originalEpoch, 0,
      new SimpleRecord("ks-initial-key".getBytes, "ks-initial-value".getBytes)
    )
    val initialGuard = log.maybeStartTransactionVerification(producerId, 0, originalEpoch, true) // TV2 = supportsEpochBump = true
    log.appendAsLeader(initialRecords, 0, AppendOrigin.CLIENT, RequestLocal.noCaching, initialGuard)

    // Step 2: Coordinator times out and aborts transaction
    // TV2 (KIP-890): Coordinator bumps epoch from 3 → 4 and sends abort marker with epoch 4
    val abortMarker = MemoryRecords.withEndTransactionMarker(
      producerId, bumpedEpoch, new EndTransactionMarker(ControlRecordType.ABORT, 0)
    )
    log.appendAsLeader(abortMarker, 0, AppendOrigin.COORDINATOR, RequestLocal.noCaching, VerificationGuard.SENTINEL)

    // Step 3: TV2 transactional producer tries to append with stale epoch (timeout recovery scenario)
    val staleEpochRecords = MemoryRecords.withTransactionalRecords(
      Compression.NONE, producerId, originalEpoch, 0,
      new SimpleRecord("ks-resume-key".getBytes, "ks-resume-value".getBytes)
    )

    // Step 4: Verify our fix works for TV2 - should get InvalidProducerEpochException (recoverable), not InvalidTxnStateException (fatal)
    val exception = assertThrows(classOf[InvalidProducerEpochException], () => {
      val staleGuard = log.maybeStartTransactionVerification(producerId, 0, originalEpoch, true) // TV2 = supportsEpochBump = true
      log.appendAsLeader(staleEpochRecords, 0, AppendOrigin.CLIENT, RequestLocal.noCaching, staleGuard)
    })

    // Verify the error message indicates epoch mismatch (3 < 4)
    assertTrue(exception.getMessage.contains("smaller than the last seen epoch"))
    assertTrue(exception.getMessage.contains(s"$originalEpoch"))
    assertTrue(exception.getMessage.contains(s"$bumpedEpoch"))
  }

  private def appendTransactionalToBuffer(buffer: ByteBuffer,
                                          producerId: Long,
                                          producerEpoch: Short,
                                          leaderEpoch: Int = 0): (Long, Int) => Unit = {
    var sequence = 0
    (offset: Long, numRecords: Int) => {
      val builder = MemoryRecords.builder(buffer, RecordBatch.CURRENT_MAGIC_VALUE, Compression.NONE, TimestampType.CREATE_TIME,
        offset, mockTime.milliseconds(), producerId, producerEpoch, sequence, true, leaderEpoch)
      for (seq <- sequence until sequence + numRecords) {
        val record = new SimpleRecord(s"$seq".getBytes)