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 3 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
107 changes: 60 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,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)

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
Comment thread
jolshan marked this conversation as resolved.
Outdated
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)

@artemlivshits artemlivshits Apr 18, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need this condition for verification case? No matter what the pending state is, if the the state contains the partition, we're good, otherwise, we'd fail during verification.

@hachikuji hachikuji Apr 19, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, I think that's right. It works because we remove the partition after we have confirmed that the end transaction marker has been written. So if the partition is included, then it means markers are still to come. This assumes we fix https://issues.apache.org/jira/browse/KAFKA-14884 of course.

@jolshan jolshan Apr 19, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we are pending commit or abort I don't know if it makes sense to verify and allow the write to continue.

@jolshan jolshan Apr 19, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we allow any pending states, we will have to catch the error on kafka-14884. Which I suppose is also ok

} else if (txnMetadata.state == PrepareCommit || txnMetadata.state == PrepareAbort) {
Left(Errors.CONCURRENT_TRANSACTIONS)
} else {
Right(partitions.map(part =>
Comment thread
jolshan marked this conversation as resolved.
Outdated
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 +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)
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
43 changes: 43 additions & 0 deletions 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,49 @@ 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
}

// 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, emptyTxnMetadata))))

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
def shouldRespondWithConcurrentTransactionsOnAddPartitionsWhenStateIsPrepareCommit(): Unit = {
Expand Down