This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new cb69d5716e5 MINOR: Rewrite TransactionMetadataTest from Scala to Java
(#22811)
cb69d5716e5 is described below
commit cb69d5716e5924001ddc269ec95ae5a09e1f80f2
Author: Ken Huang <[email protected]>
AuthorDate: Tue Jul 21 02:20:44 2026 +0800
MINOR: Rewrite TransactionMetadataTest from Scala to Java (#22811)
Rewrite it, and move it to transaction coordinator module
Reviewers: Chia-Ping Tsai <[email protected]>
---
.../transaction/TransactionMetadataTest.scala | 794 --------------------
.../transaction/TransactionMetadataTest.java | 822 +++++++++++++++++++++
2 files changed, 822 insertions(+), 794 deletions(-)
diff --git
a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMetadataTest.scala
b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMetadataTest.scala
deleted file mode 100644
index cab8d3e90de..00000000000
---
a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMetadataTest.scala
+++ /dev/null
@@ -1,794 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package kafka.coordinator.transaction
-
-import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.record.internal.RecordBatch
-import org.apache.kafka.coordinator.transaction.{TransactionMetadata,
TransactionState, TxnTransitMetadata}
-import org.apache.kafka.server.common.TransactionVersion
-import org.apache.kafka.server.common.TransactionVersion.{TV_0, TV_2}
-import org.apache.kafka.server.util.MockTime
-import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api.Test
-import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.ValueSource
-
-import java.util
-import java.util.Optional
-
-import scala.collection.mutable
-import scala.jdk.CollectionConverters._
-
-class TransactionMetadataTest {
-
- val time = new MockTime()
- val producerId = 23423L
- val transactionalId = "txnlId"
-
- @Test
- def testInitializeEpoch(): Unit = {
- val producerEpoch = RecordBatch.NO_PRODUCER_EPOCH
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.EMPTY,
- util.Set.of,
- -1,
- time.milliseconds(),
- TV_0)
-
- val transitMetadata = prepareSuccessfulIncrementProducerEpoch(txnMetadata,
Optional.empty())
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(0, txnMetadata.producerEpoch)
- assertEquals(RecordBatch.NO_PRODUCER_EPOCH, txnMetadata.lastProducerEpoch)
- }
-
- @Test
- def testNormalEpochBump(): Unit = {
- val producerEpoch = 735.toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.EMPTY,
- util.Set.of,
- -1,
- time.milliseconds(),
- TV_0)
-
- val transitMetadata = prepareSuccessfulIncrementProducerEpoch(txnMetadata,
Optional.empty())
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(producerEpoch + 1, txnMetadata.producerEpoch)
- assertEquals(RecordBatch.NO_PRODUCER_EPOCH, txnMetadata.lastProducerEpoch)
- }
-
- @Test
- def testBumpEpochNotAllowedIfEpochsExhausted(): Unit = {
- val producerEpoch = (Short.MaxValue - 1).toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.EMPTY,
- util.Set.of,
- -1,
- time.milliseconds(),
- TV_0)
- assertTrue(txnMetadata.isProducerEpochExhausted)
-
- assertThrows(classOf[IllegalStateException], () =>
txnMetadata.prepareIncrementProducerEpoch(30000,
- Optional.empty, time.milliseconds()))
- }
-
- @Test
- def testTransitFromEmptyToPrepareAbortInV2(): Unit = {
- val producerEpoch = 735.toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.EMPTY,
- util.Set.of,
- -1,
- time.milliseconds(),
- TV_2)
-
- val transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_ABORT, TV_2,
RecordBatch.NO_PRODUCER_ID, time.milliseconds() + 1, true)
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(producerEpoch + 1, txnMetadata.producerEpoch)
- assertEquals(time.milliseconds() + 1, txnMetadata.txnStartTimestamp)
- }
-
- @Test
- def testTransitFromCompleteAbortToPrepareAbortInV2(): Unit = {
- val producerEpoch = 735.toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.COMPLETE_ABORT,
- util.Set.of,
- time.milliseconds() - 1,
- time.milliseconds(),
- TV_2)
-
- val transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_ABORT, TV_2,
RecordBatch.NO_PRODUCER_ID, time.milliseconds() + 1, true)
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(producerEpoch + 1, txnMetadata.producerEpoch)
- assertEquals(time.milliseconds() + 1, txnMetadata.txnStartTimestamp)
- }
-
- @Test
- def testTransitFromCompleteCommitToPrepareAbortInV2(): Unit = {
- val producerEpoch = 735.toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.COMPLETE_COMMIT,
- util.Set.of,
- time.milliseconds() - 1,
- time.milliseconds(),
- TV_2)
-
- val transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_ABORT, TV_2,
RecordBatch.NO_PRODUCER_ID, time.milliseconds() + 1, true)
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(producerEpoch + 1, txnMetadata.producerEpoch)
- assertEquals(time.milliseconds() + 1, txnMetadata.txnStartTimestamp)
- }
-
- @Test
- def testTolerateUpdateTimeShiftDuringEpochBump(): Unit = {
- val producerEpoch: Short = 1
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.EMPTY,
- util.Set.of,
- 1L,
- time.milliseconds(),
- TV_0)
-
- // let new time be smaller
- val transitMetadata = prepareSuccessfulIncrementProducerEpoch(txnMetadata,
Optional.of(producerEpoch),
- Some(time.milliseconds() - 1))
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(producerEpoch + 1, txnMetadata.producerEpoch)
- assertEquals(producerEpoch, txnMetadata.lastProducerEpoch)
- assertEquals(-1L, txnMetadata.txnStartTimestamp)
- assertEquals(time.milliseconds() - 1, txnMetadata.txnLastUpdateTimestamp)
- }
-
- @Test
- def testTolerateUpdateTimeResetDuringProducerIdRotation(): Unit = {
- val producerEpoch: Short = 1
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.EMPTY,
- util.Set.of,
- 1L,
- time.milliseconds(),
- TV_0)
-
- // let new time be smaller
- val transitMetadata = txnMetadata.prepareProducerIdRotation(producerId +
1, 30000, time.milliseconds() - 1, true)
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId + 1, txnMetadata.producerId)
- assertEquals(producerEpoch, txnMetadata.lastProducerEpoch)
- assertEquals(0, txnMetadata.producerEpoch)
- assertEquals(-1L, txnMetadata.txnStartTimestamp)
- assertEquals(time.milliseconds() - 1, txnMetadata.txnLastUpdateTimestamp)
- }
-
- @Test
- def testTolerateTimeShiftDuringAddPartitions(): Unit = {
- val producerEpoch: Short = 1
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.EMPTY,
- util.Set.of,
- time.milliseconds(),
- time.milliseconds(),
- TV_0)
-
- // let new time be smaller; when transiting from TransactionState.EMPTY
the start time would be updated to the update-time
- var transitMetadata = txnMetadata.prepareAddPartitions(util.Set.of(new
TopicPartition("topic1", 0)), time.milliseconds() - 1, TV_0)
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(util.Set.of(new TopicPartition("topic1", 0)),
txnMetadata.topicPartitions)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(RecordBatch.NO_PRODUCER_EPOCH, txnMetadata.lastProducerEpoch)
- assertEquals(producerEpoch, txnMetadata.producerEpoch)
- assertEquals(time.milliseconds() - 1, txnMetadata.txnStartTimestamp)
- assertEquals(time.milliseconds() - 1, txnMetadata.txnLastUpdateTimestamp)
-
- // add another partition, check that in TransactionState.ONGOING state the
start timestamp would not change to update time
- transitMetadata = txnMetadata.prepareAddPartitions(util.Set.of(new
TopicPartition("topic2", 0)), time.milliseconds() - 2, TV_0)
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(util.Set.of(new TopicPartition("topic1", 0), new
TopicPartition("topic2", 0)), txnMetadata.topicPartitions)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(RecordBatch.NO_PRODUCER_EPOCH, txnMetadata.lastProducerEpoch)
- assertEquals(producerEpoch, txnMetadata.producerEpoch)
- assertEquals(time.milliseconds() - 1, txnMetadata.txnStartTimestamp)
- assertEquals(time.milliseconds() - 2, txnMetadata.txnLastUpdateTimestamp)
- }
-
- @Test
- def testTolerateTimeShiftDuringPrepareCommit(): Unit = {
- val producerEpoch: Short = 1
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.ONGOING,
- util.Set.of,
- 1L,
- time.milliseconds(),
- TV_0)
-
- // let new time be smaller
- val transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_COMMIT, TV_0,
RecordBatch.NO_PRODUCER_ID, time.milliseconds() - 1, false)
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(TransactionState.PREPARE_COMMIT, txnMetadata.state)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(RecordBatch.NO_PRODUCER_EPOCH, txnMetadata.lastProducerEpoch)
- assertEquals(producerEpoch, txnMetadata.producerEpoch)
- assertEquals(1L, txnMetadata.txnStartTimestamp)
- assertEquals(time.milliseconds() - 1, txnMetadata.txnLastUpdateTimestamp)
- }
-
- @Test
- def testTolerateTimeShiftDuringPrepareAbort(): Unit = {
- val producerEpoch: Short = 1
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.ONGOING,
- util.Set.of,
- 1L,
- time.milliseconds(),
- TV_0)
-
- // let new time be smaller
- val transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_ABORT, TV_0,
RecordBatch.NO_PRODUCER_ID, time.milliseconds() - 1, false)
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(TransactionState.PREPARE_ABORT, txnMetadata.state)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(RecordBatch.NO_PRODUCER_EPOCH, txnMetadata.lastProducerEpoch)
- assertEquals(producerEpoch, txnMetadata.producerEpoch)
- assertEquals(1L, txnMetadata.txnStartTimestamp)
- assertEquals(time.milliseconds() - 1, txnMetadata.txnLastUpdateTimestamp)
- }
-
- @ParameterizedTest
- @ValueSource(shorts = Array(0, 2))
- def testTolerateTimeShiftDuringCompleteCommit(transactionVersion: Short):
Unit = {
- val clientTransactionVersion =
TransactionVersion.fromFeatureLevel(transactionVersion)
- val producerEpoch: Short = 1
- val lastProducerEpoch: Short = 0
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- lastProducerEpoch,
- 30000,
- TransactionState.PREPARE_COMMIT,
- util.Set.of(),
- 1L,
- time.milliseconds(),
- clientTransactionVersion
- )
-
- // let new time be smaller
- val transitMetadata = txnMetadata.prepareComplete(time.milliseconds() - 1)
- txnMetadata.completeTransitionTo(transitMetadata)
-
- assertEquals(TransactionState.COMPLETE_COMMIT, txnMetadata.state)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(lastProducerEpoch, txnMetadata.lastProducerEpoch)
- assertEquals(producerEpoch, txnMetadata.producerEpoch)
- assertEquals(1L, txnMetadata.txnStartTimestamp)
- assertEquals(time.milliseconds() - 1, txnMetadata.txnLastUpdateTimestamp)
- }
-
- @ParameterizedTest
- @ValueSource(shorts = Array(0, 2))
- def testTolerateTimeShiftDuringCompleteAbort(transactionVersion: Short):
Unit = {
- val clientTransactionVersion =
TransactionVersion.fromFeatureLevel(transactionVersion)
- val producerEpoch: Short = 1
- val lastProducerEpoch: Short = 0
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- lastProducerEpoch,
- 30000,
- TransactionState.PREPARE_ABORT,
- util.Set.of,
- 1L,
- time.milliseconds(),
- clientTransactionVersion
- )
-
- // let new time be smaller
- val transitMetadata = txnMetadata.prepareComplete(time.milliseconds() - 1)
- txnMetadata.completeTransitionTo(transitMetadata)
-
- assertEquals(TransactionState.COMPLETE_ABORT, txnMetadata.state)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(lastProducerEpoch, txnMetadata.lastProducerEpoch)
- assertEquals(producerEpoch, txnMetadata.producerEpoch)
- assertEquals(1L, txnMetadata.txnStartTimestamp)
- assertEquals(time.milliseconds() - 1, txnMetadata.txnLastUpdateTimestamp)
- }
-
- @Test
- def testFenceProducerAfterEpochsExhausted(): Unit = {
- val producerEpoch = (Short.MaxValue - 1).toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.ONGOING,
- util.Set.of,
- -1,
- time.milliseconds(),
- TV_0)
- assertTrue(txnMetadata.isProducerEpochExhausted)
-
- val fencingTransitMetadata = txnMetadata.prepareFenceProducerEpoch()
- assertEquals(Short.MaxValue, fencingTransitMetadata.producerEpoch)
- assertEquals(RecordBatch.NO_PRODUCER_EPOCH,
fencingTransitMetadata.lastProducerEpoch)
- assertEquals(Optional.of(TransactionState.PREPARE_EPOCH_FENCE),
txnMetadata.pendingState)
-
- // We should reset the pending state to make way for the abort transition.
- txnMetadata.pendingState(Optional.empty())
-
- val transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_ABORT, TV_0,
RecordBatch.NO_PRODUCER_ID, time.milliseconds(), false)
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId, transitMetadata.producerId)
- }
-
- @Test
- def testInvalidTransitionFromCompleteCommitToFence(): Unit = {
- val producerEpoch = (Short.MaxValue - 1).toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.COMPLETE_COMMIT,
- util.Set.of,
- -1,
- time.milliseconds(),
- TV_0)
- assertTrue(txnMetadata.isProducerEpochExhausted)
-
- assertThrows(classOf[IllegalStateException], () =>
txnMetadata.prepareFenceProducerEpoch())
- }
-
- @Test
- def testInvalidTransitionFromCompleteAbortToFence(): Unit = {
- val producerEpoch = (Short.MaxValue - 1).toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.COMPLETE_ABORT,
- util.Set.of,
- -1,
- time.milliseconds(),
- TV_0)
- assertTrue(txnMetadata.isProducerEpochExhausted)
-
- assertThrows(classOf[IllegalStateException], () =>
txnMetadata.prepareFenceProducerEpoch())
- }
-
- @Test
- def testFenceProducerNotAllowedIfItWouldOverflow(): Unit = {
- val producerEpoch = Short.MaxValue
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.ONGOING,
- util.Set.of,
- -1,
- time.milliseconds(),
- TV_0)
- assertTrue(txnMetadata.isProducerEpochExhausted)
-
- // When epoch is at max, prepareFenceProducerEpoch logs an error but
doesn't throw
- // This allows graceful recovery through producer ID rotation
- val preparedMetadata = txnMetadata.prepareFenceProducerEpoch()
-
- // Epoch should remain at Short.MaxValue (not overflow to negative)
- assertEquals(Short.MaxValue, preparedMetadata.producerEpoch)
- assertEquals(TransactionState.PREPARE_EPOCH_FENCE,
preparedMetadata.txnState)
- }
-
- @Test
- def testRotateProducerId(): Unit = {
- val producerEpoch = (Short.MaxValue - 1).toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.EMPTY,
- util.Set.of,
- -1,
- time.milliseconds(),
- TV_0)
-
- val newProducerId = 9893L
- val transitMetadata = txnMetadata.prepareProducerIdRotation(newProducerId,
30000, time.milliseconds(), true)
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(newProducerId, txnMetadata.producerId)
- assertEquals(producerId, txnMetadata.prevProducerId)
- assertEquals(0, txnMetadata.producerEpoch)
- assertEquals(producerEpoch, txnMetadata.lastProducerEpoch)
- }
-
- @Test
- def testEpochBumpOnEndTxn(): Unit = {
- time.sleep(100)
- val producerEpoch = 10.toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.ONGOING,
- util.Set.of,
- time.milliseconds(),
- time.milliseconds(),
- TV_2)
-
- var transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_COMMIT, TV_2,
RecordBatch.NO_PRODUCER_ID, time.milliseconds() - 1, false)
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals((producerEpoch + 1).toShort, txnMetadata.producerEpoch)
- assertEquals(TV_2, txnMetadata.clientTransactionVersion)
-
- transitMetadata = txnMetadata.prepareComplete(time.milliseconds())
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals((producerEpoch + 1).toShort, txnMetadata.producerEpoch)
- assertEquals(TV_2, txnMetadata.clientTransactionVersion)
- }
-
- @Test
- def testEpochBumpOnEndTxnOverflow(): Unit = {
- time.sleep(100)
- val producerEpoch = (Short.MaxValue - 1).toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.ONGOING,
- util.Set.of,
- time.milliseconds(),
- time.milliseconds(),
- TV_2)
- assertTrue(txnMetadata.isProducerEpochExhausted)
-
- val newProducerId = 9893L
- var transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_COMMIT, TV_2,
newProducerId, time.milliseconds() - 1, false)
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(Short.MaxValue, txnMetadata.producerEpoch)
- assertEquals(producerEpoch, txnMetadata.lastProducerEpoch)
- assertEquals(TV_2, txnMetadata.clientTransactionVersion)
-
- transitMetadata = txnMetadata.prepareComplete(time.milliseconds())
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(newProducerId, txnMetadata.producerId)
- assertEquals(0, txnMetadata.producerEpoch)
- assertEquals(producerEpoch, txnMetadata.lastProducerEpoch)
- assertEquals(TV_2, txnMetadata.clientTransactionVersion)
- }
-
- @Test
- def testRotateProducerIdInOngoingState(): Unit = {
- assertThrows(classOf[IllegalStateException], () =>
testRotateProducerIdInOngoingState(TransactionState.ONGOING, TV_0))
- }
-
- @ParameterizedTest
- @ValueSource(shorts = Array(0, 2))
- def testRotateProducerIdInPrepareAbortState(transactionVersion: Short): Unit
= {
- val clientTransactionVersion =
TransactionVersion.fromFeatureLevel(transactionVersion)
- assertThrows(classOf[IllegalStateException], () =>
testRotateProducerIdInOngoingState(TransactionState.PREPARE_ABORT,
clientTransactionVersion))
- }
-
- @ParameterizedTest
- @ValueSource(shorts = Array(0, 2))
- def testRotateProducerIdInPrepareCommitState(transactionVersion: Short):
Unit = {
- val clientTransactionVersion =
TransactionVersion.fromFeatureLevel(transactionVersion)
- assertThrows(classOf[IllegalStateException], () =>
testRotateProducerIdInOngoingState(TransactionState.PREPARE_COMMIT,
clientTransactionVersion))
- }
-
- @Test
- def testAttemptedEpochBumpWithNewlyCreatedMetadata(): Unit = {
- val producerEpoch = 735.toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_EPOCH,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.EMPTY,
- util.Set.of,
- -1,
- time.milliseconds(),
- TV_0)
-
- val transitMetadata = prepareSuccessfulIncrementProducerEpoch(txnMetadata,
Optional.of(producerEpoch))
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(0, txnMetadata.producerEpoch)
- assertEquals(RecordBatch.NO_PRODUCER_EPOCH, txnMetadata.lastProducerEpoch)
- }
-
- @Test
- def testEpochBumpWithCurrentEpochProvided(): Unit = {
- val producerEpoch = 735.toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- TransactionState.EMPTY,
- util.Set.of,
- -1,
- time.milliseconds(),
- TV_0)
-
- val transitMetadata = prepareSuccessfulIncrementProducerEpoch(txnMetadata,
Optional.of(producerEpoch))
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(producerEpoch + 1, txnMetadata.producerEpoch)
- assertEquals(producerEpoch, txnMetadata.lastProducerEpoch)
- }
-
- @Test
- def testAttemptedEpochBumpWithLastEpoch(): Unit = {
- val producerEpoch = 735.toShort
- val lastProducerEpoch = (producerEpoch - 1).toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- lastProducerEpoch,
- 30000,
- TransactionState.EMPTY,
- util.Set.of,
- -1,
- time.milliseconds(),
- TV_0)
-
- val transitMetadata = prepareSuccessfulIncrementProducerEpoch(txnMetadata,
Optional.of(lastProducerEpoch))
- txnMetadata.completeTransitionTo(transitMetadata)
- assertEquals(producerId, txnMetadata.producerId)
- assertEquals(producerEpoch, txnMetadata.producerEpoch)
- assertEquals(lastProducerEpoch, txnMetadata.lastProducerEpoch)
- }
-
- @Test
- def testAttemptedEpochBumpWithFencedEpoch(): Unit = {
- val producerEpoch = 735.toShort
- val lastProducerEpoch = (producerEpoch - 1).toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- lastProducerEpoch,
- 30000,
- TransactionState.EMPTY,
- util.Set.of,
- -1,
- time.milliseconds(),
- TV_0)
-
- assertThrows(Errors.PRODUCER_FENCED.exception().getClass, () =>
- txnMetadata.prepareIncrementProducerEpoch(30000,
Optional.of((lastProducerEpoch - 1).toShort),
- time.milliseconds())
- )
- }
-
- @Test
- def testTransactionStateIdAndNameMapping(): Unit = {
- for (state <- TransactionState.ALL_STATES.asScala) {
- assertEquals(state, TransactionState.fromId(state.id))
- assertEquals(Optional.of(state),
TransactionState.fromName(state.stateName))
-
- if (state != TransactionState.DEAD) {
- val clientTransactionState =
org.apache.kafka.clients.admin.TransactionState.parse(state.stateName)
- assertEquals(state.stateName, clientTransactionState.toString)
-
assertNotEquals(org.apache.kafka.clients.admin.TransactionState.UNKNOWN,
clientTransactionState)
- }
- }
- }
-
- @Test
- def testAllTransactionStatesAreMapped(): Unit = {
- val unmatchedStates = mutable.Set(
- TransactionState.EMPTY,
- TransactionState.ONGOING,
- TransactionState.PREPARE_COMMIT,
- TransactionState.PREPARE_ABORT,
- TransactionState.COMPLETE_COMMIT,
- TransactionState.COMPLETE_ABORT,
- TransactionState.PREPARE_EPOCH_FENCE,
- TransactionState.DEAD
- )
-
- // The exhaustive match is intentional here to ensure that we are
- // forced to update the test case if a new state is added.
- TransactionState.ALL_STATES.asScala.foreach {
- case TransactionState.EMPTY =>
assertTrue(unmatchedStates.remove(TransactionState.EMPTY))
- case TransactionState.ONGOING =>
assertTrue(unmatchedStates.remove(TransactionState.ONGOING))
- case TransactionState.PREPARE_COMMIT =>
assertTrue(unmatchedStates.remove(TransactionState.PREPARE_COMMIT))
- case TransactionState.PREPARE_ABORT =>
assertTrue(unmatchedStates.remove(TransactionState.PREPARE_ABORT))
- case TransactionState.COMPLETE_COMMIT =>
assertTrue(unmatchedStates.remove(TransactionState.COMPLETE_COMMIT))
- case TransactionState.COMPLETE_ABORT =>
assertTrue(unmatchedStates.remove(TransactionState.COMPLETE_ABORT))
- case TransactionState.PREPARE_EPOCH_FENCE =>
assertTrue(unmatchedStates.remove(TransactionState.PREPARE_EPOCH_FENCE))
- case TransactionState.DEAD =>
assertTrue(unmatchedStates.remove(TransactionState.DEAD))
- }
-
- assertEquals(Set.empty, unmatchedStates)
- }
-
- private def testRotateProducerIdInOngoingState(state: TransactionState,
clientTransactionVersion: TransactionVersion): Unit = {
- val producerEpoch = (Short.MaxValue - 1).toShort
-
- val txnMetadata = new TransactionMetadata(
- transactionalId,
- producerId,
- producerId,
- RecordBatch.NO_PRODUCER_ID,
- producerEpoch,
- RecordBatch.NO_PRODUCER_EPOCH,
- 30000,
- state,
- util.Set.of,
- -1,
- time.milliseconds(),
- clientTransactionVersion)
- val newProducerId = 9893L
- txnMetadata.prepareProducerIdRotation(newProducerId, 30000,
time.milliseconds(), false)
- }
-
- private def prepareSuccessfulIncrementProducerEpoch(txnMetadata:
TransactionMetadata,
- expectedProducerEpoch:
Optional[java.lang.Short],
- now: Option[Long] =
None): TxnTransitMetadata = {
- txnMetadata.prepareIncrementProducerEpoch(30000, expectedProducerEpoch,
now.getOrElse(time.milliseconds()))
- }
-
-}
diff --git
a/transaction-coordinator/src/test/java/org/apache/kafka/coordinator/transaction/TransactionMetadataTest.java
b/transaction-coordinator/src/test/java/org/apache/kafka/coordinator/transaction/TransactionMetadataTest.java
new file mode 100644
index 00000000000..9c33f25dbfd
--- /dev/null
+++
b/transaction-coordinator/src/test/java/org/apache/kafka/coordinator/transaction/TransactionMetadataTest.java
@@ -0,0 +1,822 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.transaction;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ProducerFencedException;
+import org.apache.kafka.common.record.internal.RecordBatch;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.server.common.TransactionVersion;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.server.common.TransactionVersion.TV_0;
+import static org.apache.kafka.server.common.TransactionVersion.TV_2;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+public class TransactionMetadataTest {
+
+ private final MockTime time = new MockTime();
+ private final long producerId = 23423L;
+ private final String transactionalId = "txnlId";
+
+ @Test
+ public void testInitializeEpoch() {
+ short producerEpoch = RecordBatch.NO_PRODUCER_EPOCH;
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.EMPTY,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ TV_0
+ );
+
+ TxnTransitMetadata transitMetadata =
prepareSuccessfulIncrementProducerEpoch(txnMetadata, Optional.empty());
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(0, txnMetadata.producerEpoch());
+ assertEquals(RecordBatch.NO_PRODUCER_EPOCH,
txnMetadata.lastProducerEpoch());
+ }
+
+ @Test
+ public void testNormalEpochBump() {
+ short producerEpoch = 735;
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.EMPTY,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ TV_0
+ );
+
+ TxnTransitMetadata transitMetadata =
prepareSuccessfulIncrementProducerEpoch(txnMetadata, Optional.empty());
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(producerEpoch + 1, txnMetadata.producerEpoch());
+ assertEquals(RecordBatch.NO_PRODUCER_EPOCH,
txnMetadata.lastProducerEpoch());
+ }
+
+ @Test
+ public void testBumpEpochNotAllowedIfEpochsExhausted() {
+ short producerEpoch = (short) (Short.MAX_VALUE - 1);
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.EMPTY,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ TV_0
+ );
+
+ assertTrue(txnMetadata.isProducerEpochExhausted());
+ assertThrows(IllegalStateException.class, () ->
txnMetadata.prepareIncrementProducerEpoch(30000,
+ Optional.empty(), time.milliseconds()));
+ }
+
+ @Test
+ public void testTransitFromEmptyToPrepareAbortInV2() {
+ short producerEpoch = 735;
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.EMPTY,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ TV_2
+ );
+
+ TxnTransitMetadata transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_ABORT, TV_2,
RecordBatch.NO_PRODUCER_ID, time.milliseconds() + 1, true);
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(producerEpoch + 1, txnMetadata.producerEpoch());
+ assertEquals(time.milliseconds() + 1, txnMetadata.txnStartTimestamp());
+ }
+
+ @Test
+ public void testTransitFromCompleteAbortToPrepareAbortInV2() {
+ short producerEpoch = 735;
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.COMPLETE_ABORT,
+ Set.of(),
+ time.milliseconds() - 1,
+ time.milliseconds(),
+ TV_2
+ );
+
+ TxnTransitMetadata transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_ABORT, TV_2,
RecordBatch.NO_PRODUCER_ID, time.milliseconds() + 1, true);
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(producerEpoch + 1, txnMetadata.producerEpoch());
+ assertEquals(time.milliseconds() + 1, txnMetadata.txnStartTimestamp());
+ }
+
+ @Test
+ public void testTransitFromCompleteCommitToPrepareAbortInV2() {
+ short producerEpoch = 735;
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.COMPLETE_COMMIT,
+ Set.of(),
+ time.milliseconds() - 1,
+ time.milliseconds(),
+ TV_2
+ );
+
+ TxnTransitMetadata transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_ABORT, TV_2,
RecordBatch.NO_PRODUCER_ID, time.milliseconds() + 1, true);
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(producerEpoch + 1, txnMetadata.producerEpoch());
+ assertEquals(time.milliseconds() + 1, txnMetadata.txnStartTimestamp());
+ }
+
+ @Test
+ public void testTolerateUpdateTimeShiftDuringEpochBump() {
+ short producerEpoch = 1;
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.EMPTY,
+ Set.of(),
+ 1L,
+ time.milliseconds(),
+ TV_0
+ );
+
+ // let new time be smaller
+ TxnTransitMetadata transitMetadata =
prepareSuccessfulIncrementProducerEpoch(txnMetadata, Optional.of(producerEpoch),
+ Optional.of(time.milliseconds() - 1));
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(producerEpoch + 1, txnMetadata.producerEpoch());
+ assertEquals(producerEpoch, txnMetadata.lastProducerEpoch());
+ assertEquals(-1L, txnMetadata.txnStartTimestamp());
+ assertEquals(time.milliseconds() - 1,
txnMetadata.txnLastUpdateTimestamp());
+ }
+
+ @Test
+ public void testTolerateUpdateTimeResetDuringProducerIdRotation() {
+ short producerEpoch = 1;
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.EMPTY,
+ Set.of(),
+ 1L,
+ time.milliseconds(),
+ TV_0
+ );
+
+ // let new time be smaller
+ TxnTransitMetadata transitMetadata =
txnMetadata.prepareProducerIdRotation(producerId + 1, 30000,
time.milliseconds() - 1, true);
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId + 1, txnMetadata.producerId());
+ assertEquals(producerEpoch, txnMetadata.lastProducerEpoch());
+ assertEquals(0, txnMetadata.producerEpoch());
+ assertEquals(-1L, txnMetadata.txnStartTimestamp());
+ assertEquals(time.milliseconds() - 1,
txnMetadata.txnLastUpdateTimestamp());
+ }
+
+ @Test
+ public void testTolerateTimeShiftDuringAddPartitions() {
+ short producerEpoch = 1;
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.EMPTY,
+ Set.of(),
+ time.milliseconds(),
+ time.milliseconds(),
+ TV_0
+ );
+
+ // let new time be smaller; when transiting from
TransactionState.EMPTY the start time would be updated to the update-time
+ TxnTransitMetadata transitMetadata =
txnMetadata.prepareAddPartitions(Set.of(new TopicPartition("topic1", 0)),
time.milliseconds() - 1, TV_0);
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(Set.of(new TopicPartition("topic1", 0)),
txnMetadata.topicPartitions());
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(RecordBatch.NO_PRODUCER_EPOCH,
txnMetadata.lastProducerEpoch());
+ assertEquals(producerEpoch, txnMetadata.producerEpoch());
+ assertEquals(time.milliseconds() - 1, txnMetadata.txnStartTimestamp());
+ assertEquals(time.milliseconds() - 1,
txnMetadata.txnLastUpdateTimestamp());
+
+ // add another partition, check that in TransactionState.ONGOING state
the start timestamp would not change to update time
+ transitMetadata = txnMetadata.prepareAddPartitions(Set.of(new
TopicPartition("topic2", 0)), time.milliseconds() - 2, TV_0);
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(Set.of(new TopicPartition("topic1", 0), new
TopicPartition("topic2", 0)), txnMetadata.topicPartitions());
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(RecordBatch.NO_PRODUCER_EPOCH,
txnMetadata.lastProducerEpoch());
+ assertEquals(producerEpoch, txnMetadata.producerEpoch());
+ assertEquals(time.milliseconds() - 1, txnMetadata.txnStartTimestamp());
+ assertEquals(time.milliseconds() - 2,
txnMetadata.txnLastUpdateTimestamp());
+ }
+
+ @Test
+ public void testTolerateTimeShiftDuringPrepareCommit() {
+ short producerEpoch = 1;
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.ONGOING,
+ Set.of(),
+ 1L,
+ time.milliseconds(),
+ TV_0
+ );
+
+ // let new time be smaller
+ TxnTransitMetadata transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_COMMIT, TV_0,
RecordBatch.NO_PRODUCER_ID, time.milliseconds() - 1, false);
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(TransactionState.PREPARE_COMMIT, txnMetadata.state());
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(RecordBatch.NO_PRODUCER_EPOCH,
txnMetadata.lastProducerEpoch());
+ assertEquals(producerEpoch, txnMetadata.producerEpoch());
+ assertEquals(1L, txnMetadata.txnStartTimestamp());
+ assertEquals(time.milliseconds() - 1,
txnMetadata.txnLastUpdateTimestamp());
+ }
+
+ @Test
+ public void testTolerateTimeShiftDuringPrepareAbort() {
+ short producerEpoch = 1;
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.ONGOING,
+ Set.of(),
+ 1L,
+ time.milliseconds(),
+ TV_0
+ );
+
+ // let new time be smaller
+ TxnTransitMetadata transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_ABORT, TV_0,
RecordBatch.NO_PRODUCER_ID, time.milliseconds() - 1, false);
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(TransactionState.PREPARE_ABORT, txnMetadata.state());
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(RecordBatch.NO_PRODUCER_EPOCH,
txnMetadata.lastProducerEpoch());
+ assertEquals(producerEpoch, txnMetadata.producerEpoch());
+ assertEquals(1L, txnMetadata.txnStartTimestamp());
+ assertEquals(time.milliseconds() - 1,
txnMetadata.txnLastUpdateTimestamp());
+ }
+
+ @ParameterizedTest
+ @ValueSource(shorts = {0, 2})
+ public void testTolerateTimeShiftDuringCompleteCommit(short
transactionVersion) {
+ TransactionVersion clientTransactionVersion =
TransactionVersion.fromFeatureLevel(transactionVersion);
+ short producerEpoch = 1;
+ short lastProducerEpoch = 0;
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ lastProducerEpoch,
+ 30000,
+ TransactionState.PREPARE_COMMIT,
+ Set.of(),
+ 1L,
+ time.milliseconds(),
+ clientTransactionVersion
+ );
+
+ // let new time be smaller
+ TxnTransitMetadata transitMetadata =
txnMetadata.prepareComplete(time.milliseconds() - 1);
+ txnMetadata.completeTransitionTo(transitMetadata);
+
+ assertEquals(TransactionState.COMPLETE_COMMIT, txnMetadata.state());
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(lastProducerEpoch, txnMetadata.lastProducerEpoch());
+ assertEquals(producerEpoch, txnMetadata.producerEpoch());
+ assertEquals(1L, txnMetadata.txnStartTimestamp());
+ assertEquals(time.milliseconds() - 1,
txnMetadata.txnLastUpdateTimestamp());
+ }
+
+ @ParameterizedTest
+ @ValueSource(shorts = {0, 2})
+ public void testTolerateTimeShiftDuringCompleteAbort(short
transactionVersion) {
+ TransactionVersion clientTransactionVersion =
TransactionVersion.fromFeatureLevel(transactionVersion);
+ short producerEpoch = 1;
+ short lastProducerEpoch = 0;
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ lastProducerEpoch,
+ 30000,
+ TransactionState.PREPARE_ABORT,
+ Set.of(),
+ 1L,
+ time.milliseconds(),
+ clientTransactionVersion
+ );
+
+ // let new time be smaller
+ TxnTransitMetadata transitMetadata =
txnMetadata.prepareComplete(time.milliseconds() - 1);
+ txnMetadata.completeTransitionTo(transitMetadata);
+
+ assertEquals(TransactionState.COMPLETE_ABORT, txnMetadata.state());
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(lastProducerEpoch, txnMetadata.lastProducerEpoch());
+ assertEquals(producerEpoch, txnMetadata.producerEpoch());
+ assertEquals(1L, txnMetadata.txnStartTimestamp());
+ assertEquals(time.milliseconds() - 1,
txnMetadata.txnLastUpdateTimestamp());
+ }
+
+ @Test
+ public void testFenceProducerAfterEpochsExhausted() {
+ short producerEpoch = (short) (Short.MAX_VALUE - 1);
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.ONGOING,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ TV_0);
+ assertTrue(txnMetadata.isProducerEpochExhausted());
+
+ TxnTransitMetadata fencingTransitMetadata =
txnMetadata.prepareFenceProducerEpoch();
+ assertEquals(Short.MAX_VALUE, fencingTransitMetadata.producerEpoch());
+ assertEquals(RecordBatch.NO_PRODUCER_EPOCH,
fencingTransitMetadata.lastProducerEpoch());
+ assertEquals(Optional.of(TransactionState.PREPARE_EPOCH_FENCE),
txnMetadata.pendingState());
+
+ // We should reset the pending state to make way for the abort
transition.
+ txnMetadata.pendingState(Optional.empty());
+
+ TxnTransitMetadata transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_ABORT, TV_0,
RecordBatch.NO_PRODUCER_ID, time.milliseconds(), false);
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId, transitMetadata.producerId());
+ }
+
+ @Test
+ public void testInvalidTransitionFromCompleteCommitToFence() {
+ short producerEpoch = (short) (Short.MAX_VALUE - 1);
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.COMPLETE_COMMIT,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ TV_0);
+ assertTrue(txnMetadata.isProducerEpochExhausted());
+
+ assertThrows(IllegalStateException.class,
txnMetadata::prepareFenceProducerEpoch);
+ }
+
+ @Test
+ public void testInvalidTransitionFromCompleteAbortToFence() {
+ short producerEpoch = (short) (Short.MAX_VALUE - 1);
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.COMPLETE_ABORT,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ TV_0);
+ assertTrue(txnMetadata.isProducerEpochExhausted());
+
+ assertThrows(IllegalStateException.class,
txnMetadata::prepareFenceProducerEpoch);
+ }
+
+ @Test
+ public void testFenceProducerNotAllowedIfItWouldOverflow() {
+ short producerEpoch = Short.MAX_VALUE;
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.ONGOING,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ TV_0);
+ assertTrue(txnMetadata.isProducerEpochExhausted());
+
+ // When epoch is at max, prepareFenceProducerEpoch logs an error but
doesn't throw
+ // This allows graceful recovery through producer ID rotation
+ TxnTransitMetadata preparedMetadata =
txnMetadata.prepareFenceProducerEpoch();
+
+ // Epoch should remain at Short.MaxValue (not overflow to negative)
+ assertEquals(Short.MAX_VALUE, preparedMetadata.producerEpoch());
+ assertEquals(TransactionState.PREPARE_EPOCH_FENCE,
preparedMetadata.txnState());
+ }
+
+ @Test
+ public void testRotateProducerId() {
+ short producerEpoch = (short) (Short.MAX_VALUE - 1);
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.EMPTY,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ TV_0);
+
+ long newProducerId = 9893L;
+ TxnTransitMetadata transitMetadata =
txnMetadata.prepareProducerIdRotation(newProducerId, 30000,
time.milliseconds(), true);
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(newProducerId, txnMetadata.producerId());
+ assertEquals(producerId, txnMetadata.prevProducerId());
+ assertEquals(0, txnMetadata.producerEpoch());
+ assertEquals(producerEpoch, txnMetadata.lastProducerEpoch());
+ }
+
+ @Test
+ public void testEpochBumpOnEndTxn() {
+ time.sleep(100);
+ short producerEpoch = 10;
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.ONGOING,
+ Set.of(),
+ time.milliseconds(),
+ time.milliseconds(),
+ TV_2);
+
+ TxnTransitMetadata transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_COMMIT, TV_2,
RecordBatch.NO_PRODUCER_ID, time.milliseconds() - 1, false);
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals((short) (producerEpoch + 1), txnMetadata.producerEpoch());
+ assertEquals(TV_2, txnMetadata.clientTransactionVersion());
+
+ transitMetadata = txnMetadata.prepareComplete(time.milliseconds());
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals((short) (producerEpoch + 1), txnMetadata.producerEpoch());
+ assertEquals(TV_2, txnMetadata.clientTransactionVersion());
+ }
+
+ @Test
+ public void testEpochBumpOnEndTxnOverflow() {
+ time.sleep(100);
+ short producerEpoch = (short) (Short.MAX_VALUE - 1);
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.ONGOING,
+ Set.of(),
+ time.milliseconds(),
+ time.milliseconds(),
+ TV_2);
+ assertTrue(txnMetadata.isProducerEpochExhausted());
+
+ long newProducerId = 9893L;
+ TxnTransitMetadata transitMetadata =
txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_COMMIT, TV_2,
newProducerId, time.milliseconds() - 1, false);
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(Short.MAX_VALUE, txnMetadata.producerEpoch());
+ assertEquals(producerEpoch, txnMetadata.lastProducerEpoch());
+ assertEquals(TV_2, txnMetadata.clientTransactionVersion());
+
+ transitMetadata = txnMetadata.prepareComplete(time.milliseconds());
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(newProducerId, txnMetadata.producerId());
+ assertEquals(0, txnMetadata.producerEpoch());
+ assertEquals(producerEpoch, txnMetadata.lastProducerEpoch());
+ assertEquals(TV_2, txnMetadata.clientTransactionVersion());
+ }
+
+ @Test
+ public void testRotateProducerIdInOngoingState() {
+ assertThrows(IllegalStateException.class, () ->
testRotateProducerIdInOngoingState(TransactionState.ONGOING, TV_0));
+ }
+
+ @ParameterizedTest
+ @ValueSource(shorts = {0, 2})
+ public void testRotateProducerIdInPrepareAbortState(short
transactionVersion) {
+ TransactionVersion clientTransactionVersion =
TransactionVersion.fromFeatureLevel(transactionVersion);
+ assertThrows(IllegalStateException.class, () ->
testRotateProducerIdInOngoingState(TransactionState.PREPARE_ABORT,
clientTransactionVersion));
+ }
+
+ @ParameterizedTest
+ @ValueSource(shorts = {0, 2})
+ public void testRotateProducerIdInPrepareCommitState(short
transactionVersion) {
+ TransactionVersion clientTransactionVersion =
TransactionVersion.fromFeatureLevel(transactionVersion);
+ assertThrows(IllegalStateException.class, () ->
testRotateProducerIdInOngoingState(TransactionState.PREPARE_COMMIT,
clientTransactionVersion));
+ }
+
+ @Test
+ public void testAttemptedEpochBumpWithNewlyCreatedMetadata() {
+ short producerEpoch = 735;
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.EMPTY,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ TV_0);
+
+ TxnTransitMetadata transitMetadata =
prepareSuccessfulIncrementProducerEpoch(txnMetadata,
Optional.of(producerEpoch));
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(0, txnMetadata.producerEpoch());
+ assertEquals(RecordBatch.NO_PRODUCER_EPOCH,
txnMetadata.lastProducerEpoch());
+ }
+
+ @Test
+ public void testEpochBumpWithCurrentEpochProvided() {
+ short producerEpoch = 735;
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ TransactionState.EMPTY,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ TV_0);
+
+ TxnTransitMetadata transitMetadata =
prepareSuccessfulIncrementProducerEpoch(txnMetadata,
Optional.of(producerEpoch));
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(producerEpoch + 1, txnMetadata.producerEpoch());
+ assertEquals(producerEpoch, txnMetadata.lastProducerEpoch());
+ }
+
+ @Test
+ public void testAttemptedEpochBumpWithLastEpoch() {
+ short producerEpoch = 735;
+ short lastProducerEpoch = (short) (producerEpoch - 1);
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ lastProducerEpoch,
+ 30000,
+ TransactionState.EMPTY,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ TV_0);
+
+ TxnTransitMetadata transitMetadata =
prepareSuccessfulIncrementProducerEpoch(txnMetadata,
Optional.of(lastProducerEpoch));
+ txnMetadata.completeTransitionTo(transitMetadata);
+ assertEquals(producerId, txnMetadata.producerId());
+ assertEquals(producerEpoch, txnMetadata.producerEpoch());
+ assertEquals(lastProducerEpoch, txnMetadata.lastProducerEpoch());
+ }
+
+ @Test
+ public void testAttemptedEpochBumpWithFencedEpoch() {
+ short producerEpoch = 735;
+ short lastProducerEpoch = (short) (producerEpoch - 1);
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ lastProducerEpoch,
+ 30000,
+ TransactionState.EMPTY,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ TV_0);
+
+ assertThrows(
+ ProducerFencedException.class,
+ () -> txnMetadata.prepareIncrementProducerEpoch(30000,
Optional.of((short) (lastProducerEpoch - 1)), time.milliseconds())
+ );
+ }
+
+ @Test
+ public void testTransactionStateIdAndNameMapping() {
+ for (TransactionState state : TransactionState.ALL_STATES) {
+ assertEquals(state, TransactionState.fromId(state.id()));
+ assertEquals(Optional.of(state),
TransactionState.fromName(state.stateName()));
+
+ if (state != TransactionState.DEAD) {
+ org.apache.kafka.clients.admin.TransactionState
clientTransactionState =
+
org.apache.kafka.clients.admin.TransactionState.parse(state.stateName());
+ assertEquals(state.stateName(),
clientTransactionState.toString());
+
assertNotEquals(org.apache.kafka.clients.admin.TransactionState.UNKNOWN,
clientTransactionState);
+ }
+ }
+ }
+
+ @Test
+ public void testAllTransactionStatesAreMapped() {
+ Set<TransactionState> unmatchedStates = new HashSet<>(Set.of(
+ TransactionState.EMPTY,
+ TransactionState.ONGOING,
+ TransactionState.PREPARE_COMMIT,
+ TransactionState.PREPARE_ABORT,
+ TransactionState.COMPLETE_COMMIT,
+ TransactionState.COMPLETE_ABORT,
+ TransactionState.PREPARE_EPOCH_FENCE,
+ TransactionState.DEAD
+ ));
+
+ // The exhaustive match is intentional here to ensure that we are
+ // forced to update the test case if a new state is added.
+ for (TransactionState state : TransactionState.ALL_STATES) {
+ switch (state) {
+ case EMPTY ->
assertTrue(unmatchedStates.remove(TransactionState.EMPTY));
+ case ONGOING ->
assertTrue(unmatchedStates.remove(TransactionState.ONGOING));
+ case PREPARE_COMMIT ->
assertTrue(unmatchedStates.remove(TransactionState.PREPARE_COMMIT));
+ case PREPARE_ABORT ->
assertTrue(unmatchedStates.remove(TransactionState.PREPARE_ABORT));
+ case COMPLETE_COMMIT ->
assertTrue(unmatchedStates.remove(TransactionState.COMPLETE_COMMIT));
+ case COMPLETE_ABORT ->
assertTrue(unmatchedStates.remove(TransactionState.COMPLETE_ABORT));
+ case PREPARE_EPOCH_FENCE ->
assertTrue(unmatchedStates.remove(TransactionState.PREPARE_EPOCH_FENCE));
+ case DEAD ->
assertTrue(unmatchedStates.remove(TransactionState.DEAD));
+ default -> fail("Unexpected transaction state " + state);
+ }
+ }
+
+ assertEquals(Set.of(), unmatchedStates);
+ }
+
+ private void testRotateProducerIdInOngoingState(TransactionState state,
TransactionVersion clientTransactionVersion) {
+ short producerEpoch = (short) (Short.MAX_VALUE - 1);
+
+ TransactionMetadata txnMetadata = new TransactionMetadata(
+ transactionalId,
+ producerId,
+ producerId,
+ RecordBatch.NO_PRODUCER_ID,
+ producerEpoch,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 30000,
+ state,
+ Set.of(),
+ -1,
+ time.milliseconds(),
+ clientTransactionVersion
+ );
+ long newProducerId = 9893L;
+ txnMetadata.prepareProducerIdRotation(newProducerId, 30000,
time.milliseconds(), false);
+ }
+
+ private TxnTransitMetadata prepareSuccessfulIncrementProducerEpoch(
+ TransactionMetadata txnMetadata,
+ Optional<Short> expectedProducerEpoch
+ ) {
+ return prepareSuccessfulIncrementProducerEpoch(txnMetadata,
expectedProducerEpoch, Optional.empty());
+ }
+
+ private TxnTransitMetadata prepareSuccessfulIncrementProducerEpoch(
+ TransactionMetadata txnMetadata,
+ Optional<Short> expectedProducerEpoch,
+ Optional<Long> now
+ ) {
+ return txnMetadata.prepareIncrementProducerEpoch(30000,
expectedProducerEpoch, now.orElse(time.milliseconds()));
+ }
+}
