Skip to content

KAFKA-14904: Pending state blocked verification of transactions #13579

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jolshan merged 6 commits into from
Apr 20, 2023
Merged

KAFKA-14904: Pending state blocked verification of transactions #13579

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
0f6874e
Allow verification on pending state
jolshan Apr 14, 2023
983c427
remove todo part of comment.
jolshan Apr 17, 2023
6db3a19
Add locking back in.
jolshan Apr 18, 2023
d60d6c5
allow any pending state
jolshan Apr 19, 2023
f32d25a
fix test errors
jolshan Apr 19, 2023
f543f16
nits
jolshan Apr 19, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
106 changes: 59 additions & 47 deletions core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -332,24 +331,44 @@ 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[Map[TopicPartition, Errors]] =
txnManager.getTransactionState(transactionalId).flatMap {
case None => Left(Errors.INVALID_PRODUCER_ID_MAPPING)

case Some(epochAndMetadata) =>
val txnMetadata = epochAndMetadata.transactionMetadata

// 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)
} else if (txnMetadata.producerEpoch != producerEpoch) {
Left(Errors.PRODUCER_FENCED)
} 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,
Expand All @@ -364,51 +383,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)
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]): 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) {
// 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.
Expand Down
54 changes: 53 additions & 1 deletion core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -234,6 +234,56 @@ class TransactionCoordinatorTest {
coordinator.handleAddPartitionsToTransaction(transactionalId, 0L, 1, partitions, errorsCallback)
assertEquals(Errors.COORDINATOR_LOAD_IN_PROGRESS, error)
}

@Test
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, partitions, 0, 0)
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, partitions, 0, 0)
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 Prepare or AbortCommit, we return CONCURRENT_TRANSACTIONS
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))))

coordinator.handleVerifyPartitionsInTransaction(transactionalId, 0L, 0, partitions, verifyPartitionsInTxnCallback)
errors.foreach { case (_, error) =>
assertEquals(Errors.CONCURRENT_TRANSACTIONS, error)
}

// 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)
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.INVALID_TXN_STATE, error)
}
}

@Test
def shouldRespondWithConcurrentTransactionsOnAddPartitionsWhenStateIsPrepareCommit(): Unit = {
Expand Down Expand Up @@ -327,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))
}

Expand Down