From 0f6874e61716fd2374245e7dbb78dd30820994cf Mon Sep 17 00:00:00 2001 From: Justine Date: Fri, 14 Apr 2023 15:20:53 -0700 Subject: [PATCH 1/6] Allow verification on pending state --- .../transaction/TransactionCoordinator.scala | 17 ++++++++++++----- .../TransactionCoordinatorTest.scala | 18 ++++++++++++++++++ 2 files changed, 30 insertions(+), 5 deletions(-) diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala index 02142f938a878..7ac20fcc6da9e 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala @@ -332,7 +332,7 @@ class TransactionCoordinator(txnConfig: TransactionConfig, debug(s"Returning ${Errors.INVALID_REQUEST} error code to client for $transactionalId's AddPartitions request for verification") responseCallback(AddPartitionsToTxnResponse.resultForTransaction(transactionalId, partitions.map(_ -> Errors.INVALID_REQUEST).toMap.asJava)) } else { - val result: ApiResult[(Int, TransactionMetadata)] = getTransactionMetadata(transactionalId, producerId, producerEpoch, partitions) + val result: ApiResult[(Int, TransactionMetadata)] = getTransactionMetadata(transactionalId, producerId, producerEpoch, partitions, true) result match { case Left(err) => @@ -364,7 +364,7 @@ class TransactionCoordinator(txnConfig: TransactionConfig, } else { // try to update the transaction metadata and append the updated metadata to txn log; // if there is no such metadata treat it as invalid producerId mapping error. - val result: ApiResult[(Int, TransactionMetadata)] = getTransactionMetadata(transactionalId, producerId, producerEpoch, partitions) + val result: ApiResult[(Int, TransactionMetadata)] = getTransactionMetadata(transactionalId, producerId, producerEpoch, partitions, false) result match { case Left(err) => @@ -381,7 +381,8 @@ class TransactionCoordinator(txnConfig: TransactionConfig, private def getTransactionMetadata(transactionalId: String, producerId: Long, producerEpoch: Short, - partitions: collection.Set[TopicPartition]): ApiResult[(Int, TransactionMetadata)] = { + partitions: collection.Set[TopicPartition], + verifyOnly: Boolean): ApiResult[(Int, TransactionMetadata)] = { txnManager.getTransactionState(transactionalId).flatMap { case None => Left(Errors.INVALID_PRODUCER_ID_MAPPING) @@ -396,8 +397,14 @@ class TransactionCoordinator(txnConfig: TransactionConfig, } else if (txnMetadata.producerEpoch != producerEpoch) { Left(Errors.PRODUCER_FENCED) } else if (txnMetadata.pendingTransitionInProgress) { - // return a retriable exception to let the client backoff and retry - Left(Errors.CONCURRENT_TRANSACTIONS) + // If we are in the produce path, we want to avoid OutOfOrderSequence errors if the added partition is pending. + // TODO: Part 2 of KIP-890 will always start transactions with sequence 0, so we enforce that and avoid this workaround. + if (verifyOnly && txnMetadata.pendingState == Some(Ongoing) && partitions.subsetOf(txnMetadata.topicPartitions)) { + Left(Errors.NONE) + } else { + // return a retriable exception to let the client backoff and retry + Left(Errors.CONCURRENT_TRANSACTIONS) + } } else if (txnMetadata.state == PrepareCommit || txnMetadata.state == PrepareAbort) { Left(Errors.CONCURRENT_TRANSACTIONS) } else if (txnMetadata.state == Ongoing && partitions.subsetOf(txnMetadata.topicPartitions)) { diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala index fc84244cf2197..8d991eb066796 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala @@ -234,6 +234,24 @@ class TransactionCoordinatorTest { coordinator.handleAddPartitionsToTransaction(transactionalId, 0L, 1, partitions, errorsCallback) assertEquals(Errors.COORDINATOR_LOAD_IN_PROGRESS, error) } + + @Test + def shouldRespondWithNoErrorWhenVerifyingAndPendingStateIsOngoingWithPartition(): Unit = { + var errors: Map[TopicPartition, Errors] = Map.empty + + def verifyPartitionsInTxnCallback(result: AddPartitionsToTxnResult): Unit = { + errors = AddPartitionsToTxnResponse.errorsForTransaction(result.topicResults()).asScala.toMap + } + + val txnMetadata = new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, Empty, mutable.Set.empty, 0, 0) + txnMetadata.pendingState = Some(Ongoing) + partitions.foreach(txnMetadata.topicPartitions.add(_)) + when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) + .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, txnMetadata)))) + + coordinator.handleVerifyPartitionsInTransaction(transactionalId, 0L, 1, partitions, verifyPartitionsInTxnCallback) + assertEquals(Errors.NONE, error) + } @Test def shouldRespondWithConcurrentTransactionsOnAddPartitionsWhenStateIsPrepareCommit(): Unit = { From 983c4271b33604339a2d1f9996241338f2e6caa4 Mon Sep 17 00:00:00 2001 From: Justine Date: Mon, 17 Apr 2023 09:46:35 -0700 Subject: [PATCH 2/6] remove todo part of comment. --- .../kafka/coordinator/transaction/TransactionCoordinator.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala index 7ac20fcc6da9e..5c5ea0f152a2a 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala @@ -398,7 +398,7 @@ class TransactionCoordinator(txnConfig: TransactionConfig, Left(Errors.PRODUCER_FENCED) } else if (txnMetadata.pendingTransitionInProgress) { // If we are in the produce path, we want to avoid OutOfOrderSequence errors if the added partition is pending. - // TODO: Part 2 of KIP-890 will always start transactions with sequence 0, so we enforce that and avoid this workaround. + // Part 2 of KIP-890 will always start transactions with sequence 0, so we enforce that and avoid this workaround. if (verifyOnly && txnMetadata.pendingState == Some(Ongoing) && partitions.subsetOf(txnMetadata.topicPartitions)) { Left(Errors.NONE) } else { From 6db3a191965f0b37f47c3e745822150a9775b591 Mon Sep 17 00:00:00 2001 From: Justine Date: Tue, 18 Apr 2023 15:49:23 -0700 Subject: [PATCH 3/6] Add locking back in. --- .../transaction/TransactionCoordinator.scala | 114 +++++++++--------- .../TransactionCoordinatorTest.scala | 37 +++++- 2 files changed, 91 insertions(+), 60 deletions(-) diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala index 5c5ea0f152a2a..c026244579418 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala @@ -31,7 +31,6 @@ import org.apache.kafka.common.requests.{AddPartitionsToTxnResponse, Transaction import org.apache.kafka.common.utils.{LogContext, ProducerIdAndEpoch, Time} import org.apache.kafka.server.util.Scheduler -import scala.collection.mutable import scala.jdk.CollectionConverters._ object TransactionCoordinator { @@ -332,24 +331,45 @@ class TransactionCoordinator(txnConfig: TransactionConfig, debug(s"Returning ${Errors.INVALID_REQUEST} error code to client for $transactionalId's AddPartitions request for verification") responseCallback(AddPartitionsToTxnResponse.resultForTransaction(transactionalId, partitions.map(_ -> Errors.INVALID_REQUEST).toMap.asJava)) } else { - val result: ApiResult[(Int, TransactionMetadata)] = getTransactionMetadata(transactionalId, producerId, producerEpoch, partitions, true) - + val result: ApiResult[Map[TopicPartition, Errors]] = + txnManager.getTransactionState(transactionalId).flatMap { + case None => Left(Errors.INVALID_PRODUCER_ID_MAPPING) + + case Some(epochAndMetadata) => + val txnMetadata = epochAndMetadata.transactionMetadata + + // generate the new transaction metadata with added partitions + txnMetadata.inLock { + if (txnMetadata.producerId != producerId) { + Left(Errors.INVALID_PRODUCER_ID_MAPPING) + } else if (txnMetadata.producerEpoch != producerEpoch) { + Left(Errors.PRODUCER_FENCED) + } else if (txnMetadata.pendingTransitionInProgress && !(txnMetadata.pendingState == Some(Ongoing) && txnMetadata.state == Ongoing)) { + // return a retriable exception to let the client backoff and retry + Left(Errors.CONCURRENT_TRANSACTIONS) + } else if (txnMetadata.state == PrepareCommit || txnMetadata.state == PrepareAbort) { + Left(Errors.CONCURRENT_TRANSACTIONS) + } else { + Right(partitions.map(part => + if (txnMetadata.topicPartitions.contains(part)) + (part, Errors.NONE) + else + (part, Errors.INVALID_TXN_STATE) + ).toMap) + } + } + } + result match { case Left(err) => debug(s"Returning $err error code to client for $transactionalId's AddPartitions request for verification") responseCallback(AddPartitionsToTxnResponse.resultForTransaction(transactionalId, partitions.map(_ -> err).toMap.asJava)) - - case Right((_, txnMetadata)) => - val errors = mutable.Map[TopicPartition, Errors]() - partitions.foreach { tp => - if (txnMetadata.topicPartitions.contains(tp)) - errors.put(tp, Errors.NONE) - else - errors.put(tp, Errors.INVALID_TXN_STATE) - } + + case Right(errors) => responseCallback(AddPartitionsToTxnResponse.resultForTransaction(transactionalId, errors.asJava)) } } + } def handleAddPartitionsToTransaction(transactionalId: String, @@ -364,58 +384,44 @@ class TransactionCoordinator(txnConfig: TransactionConfig, } else { // try to update the transaction metadata and append the updated metadata to txn log; // if there is no such metadata treat it as invalid producerId mapping error. - val result: ApiResult[(Int, TransactionMetadata)] = getTransactionMetadata(transactionalId, producerId, producerEpoch, partitions, false) + val result: ApiResult[(Int, TxnTransitMetadata)] = txnManager.getTransactionState(transactionalId).flatMap { + case None => Left(Errors.INVALID_PRODUCER_ID_MAPPING) + + case Some(epochAndMetadata) => + val coordinatorEpoch = epochAndMetadata.coordinatorEpoch + val txnMetadata = epochAndMetadata.transactionMetadata + + // generate the new transaction metadata with added partitions + txnMetadata.inLock { + if (txnMetadata.producerId != producerId) { + Left(Errors.INVALID_PRODUCER_ID_MAPPING) + } else if (txnMetadata.producerEpoch != producerEpoch) { + Left(Errors.PRODUCER_FENCED) + } else if (txnMetadata.pendingTransitionInProgress) { + // return a retriable exception to let the client backoff and retry + Left(Errors.CONCURRENT_TRANSACTIONS) + } else if (txnMetadata.state == PrepareCommit || txnMetadata.state == PrepareAbort) { + Left(Errors.CONCURRENT_TRANSACTIONS) + } else if (txnMetadata.state == Ongoing && partitions.subsetOf(txnMetadata.topicPartitions)) { + // this is an optimization: if the partitions are already in the metadata reply OK immediately + Left(Errors.NONE) + } else { + Right(coordinatorEpoch, txnMetadata.prepareAddPartitions(partitions.toSet, time.milliseconds())) + } + } + } result match { case Left(err) => debug(s"Returning $err error code to client for $transactionalId's AddPartitions request") responseCallback(err) - case Right((coordinatorEpoch, txnMetadata)) => - txnManager.appendTransactionToLog(transactionalId, coordinatorEpoch, txnMetadata.prepareAddPartitions(partitions.toSet, time.milliseconds()), + case Right((coordinatorEpoch, newMetadata)) => + txnManager.appendTransactionToLog(transactionalId, coordinatorEpoch, newMetadata, responseCallback, requestLocal = requestLocal) } } } - - private def getTransactionMetadata(transactionalId: String, - producerId: Long, - producerEpoch: Short, - partitions: collection.Set[TopicPartition], - verifyOnly: Boolean): ApiResult[(Int, TransactionMetadata)] = { - txnManager.getTransactionState(transactionalId).flatMap { - case None => Left(Errors.INVALID_PRODUCER_ID_MAPPING) - - case Some(epochAndMetadata) => - val coordinatorEpoch = epochAndMetadata.coordinatorEpoch - val txnMetadata = epochAndMetadata.transactionMetadata - - // generate the new transaction metadata with added partitions - txnMetadata.inLock { - if (txnMetadata.producerId != producerId) { - Left(Errors.INVALID_PRODUCER_ID_MAPPING) - } else if (txnMetadata.producerEpoch != producerEpoch) { - Left(Errors.PRODUCER_FENCED) - } else if (txnMetadata.pendingTransitionInProgress) { - // If we are in the produce path, we want to avoid OutOfOrderSequence errors if the added partition is pending. - // Part 2 of KIP-890 will always start transactions with sequence 0, so we enforce that and avoid this workaround. - if (verifyOnly && txnMetadata.pendingState == Some(Ongoing) && partitions.subsetOf(txnMetadata.topicPartitions)) { - Left(Errors.NONE) - } else { - // return a retriable exception to let the client backoff and retry - Left(Errors.CONCURRENT_TRANSACTIONS) - } - } else if (txnMetadata.state == PrepareCommit || txnMetadata.state == PrepareAbort) { - Left(Errors.CONCURRENT_TRANSACTIONS) - } else if (txnMetadata.state == Ongoing && partitions.subsetOf(txnMetadata.topicPartitions)) { - // this is an optimization: if the partitions are already in the metadata reply OK immediately - Left(Errors.NONE) - } else { - Right(coordinatorEpoch, txnMetadata) - } - } - } - } /** * Load state from the given partition and begin handling requests for groups which map to this partition. diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala index 8d991eb066796..15e4b1a2cf9fd 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala @@ -243,14 +243,39 @@ class TransactionCoordinatorTest { errors = AddPartitionsToTxnResponse.errorsForTransaction(result.topicResults()).asScala.toMap } - val txnMetadata = new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, Empty, mutable.Set.empty, 0, 0) - txnMetadata.pendingState = Some(Ongoing) - partitions.foreach(txnMetadata.topicPartitions.add(_)) + // If the txn state is empty, we get CONCURRENT_TRANSACTIONS + val emptyTxnMetadata = new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, Empty, mutable.Set.empty, 0, 0) + emptyTxnMetadata.pendingState = Some(Ongoing) + partitions.foreach(emptyTxnMetadata.topicPartitions.add(_)) when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) - .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, txnMetadata)))) + .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, emptyTxnMetadata)))) - coordinator.handleVerifyPartitionsInTransaction(transactionalId, 0L, 1, partitions, verifyPartitionsInTxnCallback) - assertEquals(Errors.NONE, error) + coordinator.handleVerifyPartitionsInTransaction(transactionalId, 0L, 0, partitions, verifyPartitionsInTxnCallback) + errors.foreach { case (_, error) => + assertEquals(Errors.CONCURRENT_TRANSACTIONS, error) + } + + // If the txn state is Ongoing, but pending state is not, we get CONCURRENT_TRANSACTIONS + val ongoingTxnMetadata = new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, Ongoing, mutable.Set.empty, 0, 0) + ongoingTxnMetadata.pendingState = Some(CompleteCommit) + partitions.foreach(ongoingTxnMetadata.topicPartitions.add(_)) + when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) + .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, ongoingTxnMetadata)))) + + coordinator.handleVerifyPartitionsInTransaction(transactionalId, 0L, 0, partitions, verifyPartitionsInTxnCallback) + errors.foreach { case (_, error) => + assertEquals(Errors.CONCURRENT_TRANSACTIONS, error) + } + + // If pending state is ongoing, we can verify with the partitions already added. + ongoingTxnMetadata.pendingState = Some(Ongoing) + when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) + .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, ongoingTxnMetadata)))) + + coordinator.handleVerifyPartitionsInTransaction(transactionalId, 0L, 0, partitions, verifyPartitionsInTxnCallback) + errors.foreach { case (_, error) => + assertEquals(Errors.NONE, error) + } } @Test From d60d6c5fc8a11841ab91e716ecfa269d9f8970d0 Mon Sep 17 00:00:00 2001 From: Justine Date: Tue, 18 Apr 2023 17:37:11 -0700 Subject: [PATCH 4/6] allow any pending state --- .../transaction/TransactionCoordinator.scala | 3 -- .../TransactionCoordinatorTest.scala | 41 ++++++++++++------- 2 files changed, 26 insertions(+), 18 deletions(-) diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala index c026244579418..4c6777587bfbc 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala @@ -344,9 +344,6 @@ class TransactionCoordinator(txnConfig: TransactionConfig, Left(Errors.INVALID_PRODUCER_ID_MAPPING) } else if (txnMetadata.producerEpoch != producerEpoch) { Left(Errors.PRODUCER_FENCED) - } else if (txnMetadata.pendingTransitionInProgress && !(txnMetadata.pendingState == Some(Ongoing) && txnMetadata.state == Ongoing)) { - // return a retriable exception to let the client backoff and retry - Left(Errors.CONCURRENT_TRANSACTIONS) } else if (txnMetadata.state == PrepareCommit || txnMetadata.state == PrepareAbort) { Left(Errors.CONCURRENT_TRANSACTIONS) } else { diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala index 15e4b1a2cf9fd..61b9a72636ee2 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala @@ -236,16 +236,37 @@ class TransactionCoordinatorTest { } @Test - def shouldRespondWithNoErrorWhenVerifyingAndPendingStateIsOngoingWithPartition(): Unit = { + def testVerifyPartitionHandling(): Unit = { var errors: Map[TopicPartition, Errors] = Map.empty def verifyPartitionsInTxnCallback(result: AddPartitionsToTxnResult): Unit = { errors = AddPartitionsToTxnResponse.errorsForTransaction(result.topicResults()).asScala.toMap } + // If producer ID is not the same, return INVALID_PRODUCER_ID_MAPPING + val wrongPidTxnMetadata = new TransactionMetadata(transactionalId, 1, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, PrepareCommit, mutable.Set.empty, 0, 0) + partitions.foreach(wrongPidTxnMetadata.topicPartitions.add(_)) + when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) + .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, wrongPidTxnMetadata)))) + + coordinator.handleVerifyPartitionsInTransaction(transactionalId, 0L, 0, partitions, verifyPartitionsInTxnCallback) + errors.foreach { case (_, error) => + assertEquals(Errors.INVALID_PRODUCER_ID_MAPPING, error) + } + + + // If producer epoch is not equal, return PRODUCER_FENCED + val oldEpochTxnMetadata = new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, PrepareCommit, mutable.Set.empty, 0, 0) + partitions.foreach(oldEpochTxnMetadata.topicPartitions.add(_)) + when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) + .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, oldEpochTxnMetadata)))) + + coordinator.handleVerifyPartitionsInTransaction(transactionalId, 0L, 1, partitions, verifyPartitionsInTxnCallback) + errors.foreach { case (_, error) => + assertEquals(Errors.PRODUCER_FENCED, error) + } - // If the txn state is empty, we get CONCURRENT_TRANSACTIONS - val emptyTxnMetadata = new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, Empty, mutable.Set.empty, 0, 0) - emptyTxnMetadata.pendingState = Some(Ongoing) + // If the txn state is Prepare or AbortCommit, we return CONCURRENT_TRANSACTIONS + val emptyTxnMetadata = new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, PrepareCommit, mutable.Set.empty, 0, 0) partitions.foreach(emptyTxnMetadata.topicPartitions.add(_)) when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, emptyTxnMetadata)))) @@ -255,23 +276,13 @@ class TransactionCoordinatorTest { assertEquals(Errors.CONCURRENT_TRANSACTIONS, error) } - // If the txn state is Ongoing, but pending state is not, we get CONCURRENT_TRANSACTIONS + // If the txn state is Ongoing, it doesn't matter what pending state is. val ongoingTxnMetadata = new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, Ongoing, mutable.Set.empty, 0, 0) ongoingTxnMetadata.pendingState = Some(CompleteCommit) partitions.foreach(ongoingTxnMetadata.topicPartitions.add(_)) when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, ongoingTxnMetadata)))) - coordinator.handleVerifyPartitionsInTransaction(transactionalId, 0L, 0, partitions, verifyPartitionsInTxnCallback) - errors.foreach { case (_, error) => - assertEquals(Errors.CONCURRENT_TRANSACTIONS, error) - } - - // If pending state is ongoing, we can verify with the partitions already added. - ongoingTxnMetadata.pendingState = Some(Ongoing) - when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) - .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, ongoingTxnMetadata)))) - coordinator.handleVerifyPartitionsInTransaction(transactionalId, 0L, 0, partitions, verifyPartitionsInTxnCallback) errors.foreach { case (_, error) => assertEquals(Errors.NONE, error) From f32d25a7e0b03ffd6bb33865aace2451583ffc22 Mon Sep 17 00:00:00 2001 From: Justine Date: Tue, 18 Apr 2023 17:50:20 -0700 Subject: [PATCH 5/6] fix test errors --- .../TransactionCoordinatorTest.scala | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala index 61b9a72636ee2..ab8e1052f9385 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala @@ -243,8 +243,7 @@ class TransactionCoordinatorTest { errors = AddPartitionsToTxnResponse.errorsForTransaction(result.topicResults()).asScala.toMap } // If producer ID is not the same, return INVALID_PRODUCER_ID_MAPPING - val wrongPidTxnMetadata = new TransactionMetadata(transactionalId, 1, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, PrepareCommit, mutable.Set.empty, 0, 0) - partitions.foreach(wrongPidTxnMetadata.topicPartitions.add(_)) + val wrongPidTxnMetadata = new TransactionMetadata(transactionalId, 1, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, PrepareCommit, partitions, 0, 0) when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, wrongPidTxnMetadata)))) @@ -255,8 +254,7 @@ class TransactionCoordinatorTest { // If producer epoch is not equal, return PRODUCER_FENCED - val oldEpochTxnMetadata = new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, PrepareCommit, mutable.Set.empty, 0, 0) - partitions.foreach(oldEpochTxnMetadata.topicPartitions.add(_)) + val oldEpochTxnMetadata = new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, PrepareCommit, partitions, 0, 0) when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, oldEpochTxnMetadata)))) @@ -266,8 +264,7 @@ class TransactionCoordinatorTest { } // If the txn state is Prepare or AbortCommit, we return CONCURRENT_TRANSACTIONS - val emptyTxnMetadata = new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, PrepareCommit, mutable.Set.empty, 0, 0) - partitions.foreach(emptyTxnMetadata.topicPartitions.add(_)) + val emptyTxnMetadata = new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, PrepareCommit, partitions, 0, 0) when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, emptyTxnMetadata)))) @@ -276,16 +273,15 @@ class TransactionCoordinatorTest { assertEquals(Errors.CONCURRENT_TRANSACTIONS, error) } - // If the txn state is Ongoing, it doesn't matter what pending state is. + // Pending state does not matter, we will just check if the partitions are in the txnMetadata. val ongoingTxnMetadata = new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, Ongoing, mutable.Set.empty, 0, 0) ongoingTxnMetadata.pendingState = Some(CompleteCommit) - partitions.foreach(ongoingTxnMetadata.topicPartitions.add(_)) when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, ongoingTxnMetadata)))) coordinator.handleVerifyPartitionsInTransaction(transactionalId, 0L, 0, partitions, verifyPartitionsInTxnCallback) errors.foreach { case (_, error) => - assertEquals(Errors.NONE, error) + assertEquals(Errors.INVALID_TXN_STATE, error) } } @@ -381,7 +377,9 @@ class TransactionCoordinatorTest { new TransactionMetadata(transactionalId, 0, 0, 0, RecordBatch.NO_PRODUCER_EPOCH, 0, Ongoing, partitions, 0, 0))))) coordinator.handleVerifyPartitionsInTransaction(transactionalId, 0L, 0, partitions, verifyPartitionsInTxnCallback) - assertEquals(Errors.NONE, error) + errors.foreach { case (_, error) => + assertEquals(Errors.NONE, error) + } verify(transactionManager).getTransactionState(ArgumentMatchers.eq(transactionalId)) } From f543f16f2a545f8fbb7a7d10de43d74226cd455a Mon Sep 17 00:00:00 2001 From: Justine Date: Tue, 18 Apr 2023 17:57:50 -0700 Subject: [PATCH 6/6] nits --- .../coordinator/transaction/TransactionCoordinator.scala | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala index 4c6777587bfbc..bb1b3792c830b 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala @@ -338,7 +338,9 @@ class TransactionCoordinator(txnConfig: TransactionConfig, case Some(epochAndMetadata) => val txnMetadata = epochAndMetadata.transactionMetadata - // generate the new transaction metadata with added partitions + // Given the txnMetadata is valid, we check if the partitions are in the transaction. + // Pending state is not checked since there is a final validation on the append to the log. + // Partitions are added to metadata when the add partitions state is persisted, and removed when the end marker is persisted. txnMetadata.inLock { if (txnMetadata.producerId != producerId) { Left(Errors.INVALID_PRODUCER_ID_MAPPING) @@ -347,12 +349,12 @@ class TransactionCoordinator(txnConfig: TransactionConfig, } else if (txnMetadata.state == PrepareCommit || txnMetadata.state == PrepareAbort) { Left(Errors.CONCURRENT_TRANSACTIONS) } else { - Right(partitions.map(part => + Right(partitions.map { part => if (txnMetadata.topicPartitions.contains(part)) (part, Errors.NONE) else (part, Errors.INVALID_TXN_STATE) - ).toMap) + }.toMap) } } }