Repository: apache/kafka
File: core/src/test/scala/unit/kafka/server/TxnOffsetCommitRequestTest.scala (Scala)

Code before:
            .setPartitionIndexes(List[Integer](partition).asJava)
        ).asJava),
      requireStable = true,
      version = 9
    )
    val topicRecord = groupIdRecord.topics.asScala.find(_.name == topic).head
    val partitionRecord = topicRecord.partitions.asScala.find(_.partitionIndex == partition).head
    partitionRecord.committedOffset
  }

  private def testDelayedTxnOffsetCommitWithBumpedEpochIsRejected(useNewProtocol: Boolean): Unit = {
    val topic = "topic"
    val partition = 0
    val transactionalId = "txn"
    val groupId = "group"
    val offset = 100L

    // Creates the __consumer_offsets and __transaction_state topics because it won't be created automatically
    // in this test because it does not use FindCoordinator API.
    createOffsetsTopic()
    createTransactionStateTopic()

    // Join the consumer group. Note that we don't heartbeat here so we must use
    // a session long enough for the duration of the test.
    val (memberId: String, memberEpoch: Int) = joinConsumerGroup(groupId, useNewProtocol)
    assertTrue(memberId != JoinGroupRequest.UNKNOWN_MEMBER_ID)
    assertTrue(memberEpoch != JoinGroupRequest.UNKNOWN_GENERATION_ID)

    createTopic(topic, 1)

    for (version <- ApiKeys.TXN_OFFSET_COMMIT.oldestVersion to ApiKeys.TXN_OFFSET_COMMIT.latestVersion(isUnstableApiEnabled)) {
      val useTV2 = version >= 5

      // Initialize producer. Wait until the coordinator finishes loading.
      var producerIdAndEpoch: ProducerIdAndEpoch = null
      TestUtils.waitUntilTrue(() =>
        try {
          producerIdAndEpoch = initProducerId(
            transactionalId = transactionalId,
            producerIdAndEpoch = ProducerIdAndEpoch.NONE,
            expectedError = Errors.NONE
          )
          true
        } catch {
          case _: Throwable => false
        }, "initProducerId request failed"
      )

      addOffsetsToTxn(
        groupId = groupId,
        producerId = producerIdAndEpoch.producerId,

Review comment (suggestion):
```java
    assertNotEquals(JoinGroupRequest.UNKNOWN_MEMBER_ID, memberId)
    assertNotEquals(JoinGroupRequest.UNKNOWN_GENERATION_ID, memberEpoch)
```
`assertNotEquals` could show more readable messages.

Code after:
    val (memberId: String, memberEpoch: Int) = joinConsumerGroup(groupId, useNewProtocol)
    assertNotEquals(JoinGroupRequest.UNKNOWN_MEMBER_ID, memberId)
    assertNotEquals(JoinGroupRequest.UNKNOWN_GENERATION_ID, memberEpoch)

    createTopic(topic, 1)

    for (version <- ApiKeys.TXN_OFFSET_COMMIT.oldestVersion to ApiKeys.TXN_OFFSET_COMMIT.latestVersion(isUnstableApiEnabled)) {
      val useTV2 = version > EndTxnRequest.LAST_STABLE_VERSION_BEFORE_TRANSACTION_V2

      // Initialize producer. Wait until the coordinator finishes loading.
      var producerIdAndEpoch: ProducerIdAndEpoch = null
      TestUtils.waitUntilTrue(() =>
        try {
          producerIdAndEpoch = initProducerId(
            transactionalId = transactionalId,
            producerIdAndEpoch = ProducerIdAndEpoch.NONE,
            expectedError = Errors.NONE
          )
          true
        } catch {
          case _: Throwable => false
        }, "initProducerId request failed"
      )

      addOffsetsToTxn(
        groupId = groupId,
        producerId = producerIdAndEpoch.producerId,
        producerEpoch = producerIdAndEpoch.epoch,
        transactionalId = transactionalId
      )

      // Complete the transaction.
      endTxn(
        producerId = producerIdAndEpoch.producerId,
        producerEpoch = producerIdAndEpoch.epoch,
        transactionalId = transactionalId,
        isTransactionV2Enabled = useTV2,
        committed = true,
        expectedError = Errors.NONE
      )

      // Start a new transaction. Wait for the previous transaction to complete.
      TestUtils.waitUntilTrue(() =>
        try {
          addOffsetsToTxn(
            groupId = groupId,
            producerId = producerIdAndEpoch.producerId,
            producerEpoch = if (useTV2) (producerIdAndEpoch.epoch + 1).toShort else producerIdAndEpoch.epoch,
            transactionalId = transactionalId
          )
          true