You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "artemlivshits (via GitHub)" <gi...@apache.org> on 2024/03/12 18:23:45 UTC
[PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
artemlivshits opened a new pull request, #15524:
URL: https://github.com/apache/kafka/pull/15524
…tate
Now the removal of entries from the transactionsWithPendingMarkers map checks the value and all pending marker operations keep the value along with the operation state. This way, the pending marker operation can only delete the state it created and wouldn't accidentally delete the state from a different epoch (which could lead to "stuck" transactions).
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on PR #15524:
URL: https://github.com/apache/kafka/pull/15524#issuecomment-2001771301
LogDirFailureTest > testIOExceptionDuringLogRoll(String) > testIOExceptionDuringLogRoll(String).quorum=kraft is failing locally, will investigate.
ReplicaManagerTest -- all tests passed locally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on PR #15524:
URL: https://github.com/apache/kafka/pull/15524#issuecomment-2004953080
LogDirFailureTest passed locally as well (probably just a flake) and it doesn't seem to use any of the code that I touched, so unrelated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #15524:
URL: https://github.com/apache/kafka/pull/15524#issuecomment-2005610376
I am seeing it fail more often on your branch (for a few runs), but after merging with trunk, it seemed better. I will go ahead and merge.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #15524:
URL: https://github.com/apache/kafka/pull/15524#discussion_r1524022035
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala:
##########
@@ -109,23 +109,30 @@ object TransactionMarkerChannelManager {
}
-class TxnMarkerQueue(@volatile var destination: Node) {
+class TxnMarkerQueue(@volatile var destination: Node) extends Logging {
// keep track of the requests per txn topic partition so we can easily clear the queue
// during partition emigration
- private val markersPerTxnTopicPartition = new ConcurrentHashMap[Int, BlockingQueue[TxnIdAndMarkerEntry]]().asScala
+ private val markersPerTxnTopicPartition = new ConcurrentHashMap[Int, BlockingQueue[PendingCompleteTxnAndMarkerEntry]]().asScala
- def removeMarkersForTxnTopicPartition(partition: Int): Option[BlockingQueue[TxnIdAndMarkerEntry]] = {
+ def removeMarkersForTxnTopicPartition(partition: Int): Option[BlockingQueue[PendingCompleteTxnAndMarkerEntry]] = {
markersPerTxnTopicPartition.remove(partition)
}
- def addMarkers(txnTopicPartition: Int, txnIdAndMarker: TxnIdAndMarkerEntry): Unit = {
- val queue = CoreUtils.atomicGetOrUpdate(markersPerTxnTopicPartition, txnTopicPartition,
- new LinkedBlockingQueue[TxnIdAndMarkerEntry]())
- queue.add(txnIdAndMarker)
+ def addMarkers(txnTopicPartition: Int, pendingCompleteTxnAndMarker: PendingCompleteTxnAndMarkerEntry): Unit = {
+ val queue = CoreUtils.atomicGetOrUpdate(markersPerTxnTopicPartition, txnTopicPartition, {
+ info(s"Creating new marker queue for txn partition $txnTopicPartition to destination broker ${destination.id}")
+ new LinkedBlockingQueue[PendingCompleteTxnAndMarkerEntry]()
+ })
+ queue.add(pendingCompleteTxnAndMarker)
+
+ if (markersPerTxnTopicPartition.get(txnTopicPartition).orNull != queue) {
+ // This could happen if the queue got removed concurrently.
Review Comment:
As far as I can see, it shouldn't affect the user visible behavior. It does create an interesting state when the queue is removed in removeMarkersForTxnTopicPartition -- we could have:
1. [addMarkers] Retrieve queue.
2. [removeMarkersForTxnTopicPartition] Remove queue.
3. [removeMarkersForTxnTopicPartition] Iterate over queue, but not removeMarkersForTxn because queue is empty.
4. [addMarkers] Add markers to the queue.
Now we've effectively removed the markers while transactionsWithPendingMarkers has an entry.
This state could last for a while if the removal happened on unload (and technically the txn id could expire or etc. so this state may stay indefinitely until broker restart), but as soon as real workflow happens on this txn id that sends out markers, the proper entry will be created and the actual functionality will work as expected.
In other words, this race can lead to an orphan entry in transactionsWithPendingMarkers, but it doesn't affect anything (other than leaking a small amount of memory) until the markers are sent, and sending markers will fix it.
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerRequestCompletionHandler.scala:
##########
@@ -90,9 +90,10 @@ class TransactionMarkerRequestCompletionHandler(brokerId: Int,
val writeTxnMarkerResponse = response.responseBody.asInstanceOf[WriteTxnMarkersResponse]
val responseErrors = writeTxnMarkerResponse.errorsByProducerId
- for (txnIdAndMarker <- txnIdAndMarkerEntries.asScala) {
Review Comment:
I agree this code could benefit from some refactor, we should probably structure it so that instead of branching on wasDisconnected at the top, it should just iterate over the pending entries and check wasDisconnected in specific cases. But I think it should be done separately, as this change is fairly mechanical, that simplifies the review of what it does.
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala:
##########
@@ -109,23 +109,30 @@ object TransactionMarkerChannelManager {
}
-class TxnMarkerQueue(@volatile var destination: Node) {
+class TxnMarkerQueue(@volatile var destination: Node) extends Logging {
// keep track of the requests per txn topic partition so we can easily clear the queue
// during partition emigration
- private val markersPerTxnTopicPartition = new ConcurrentHashMap[Int, BlockingQueue[TxnIdAndMarkerEntry]]().asScala
+ private val markersPerTxnTopicPartition = new ConcurrentHashMap[Int, BlockingQueue[PendingCompleteTxnAndMarkerEntry]]().asScala
- def removeMarkersForTxnTopicPartition(partition: Int): Option[BlockingQueue[TxnIdAndMarkerEntry]] = {
+ def removeMarkersForTxnTopicPartition(partition: Int): Option[BlockingQueue[PendingCompleteTxnAndMarkerEntry]] = {
markersPerTxnTopicPartition.remove(partition)
}
- def addMarkers(txnTopicPartition: Int, txnIdAndMarker: TxnIdAndMarkerEntry): Unit = {
- val queue = CoreUtils.atomicGetOrUpdate(markersPerTxnTopicPartition, txnTopicPartition,
- new LinkedBlockingQueue[TxnIdAndMarkerEntry]())
- queue.add(txnIdAndMarker)
+ def addMarkers(txnTopicPartition: Int, pendingCompleteTxnAndMarker: PendingCompleteTxnAndMarkerEntry): Unit = {
+ val queue = CoreUtils.atomicGetOrUpdate(markersPerTxnTopicPartition, txnTopicPartition, {
+ info(s"Creating new marker queue for txn partition $txnTopicPartition to destination broker ${destination.id}")
Review Comment:
That's correct, this is logged when the value is created, a more precise logic would be to make atomicGetOrUpdate return an indication if the value is actually created (then we'd log it exactly once), but I didn't think it would be worth the complexity -- at worst we'd get a couple logs at the same time and one of them would create the queue. I'll add a comment.
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala:
##########
@@ -419,25 +432,34 @@ class TransactionMarkerChannelManager(
def removeMarkersForTxnTopicPartition(txnTopicPartitionId: Int): Unit = {
markersQueueForUnknownBroker.removeMarkersForTxnTopicPartition(txnTopicPartitionId).foreach { queue =>
- for (entry: TxnIdAndMarkerEntry <- queue.asScala)
- removeMarkersForTxnId(entry.txnId)
+ for (entry <- queue.asScala) {
Review Comment:
In cases I investigated it was a couple dozens or so, but I don't have the precise stats from a large selection of cases. We log messages on every retry of failed marker send, I think this message would be much less spammy than that (these would happen only when partitions are changed).
The disadvantage of having a single log is that it'll get eventually truncated, but these logs can help to see interesting transitions that could help to investigate race conditions related to load / unload.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15524:
URL: https://github.com/apache/kafka/pull/15524#discussion_r1526958117
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala:
##########
@@ -177,6 +178,86 @@ class TransactionMarkerChannelManagerTest {
any())
}
+ @Test
+ def shouldNotLoseTxnCompletionAfterLoad(): Unit = {
+ mockCache()
+
+ val expectedTransition = txnMetadata2.prepareComplete(time.milliseconds())
+
+ when(metadataCache.getPartitionLeaderEndpoint(
+ ArgumentMatchers.eq(partition1.topic),
+ ArgumentMatchers.eq(partition1.partition),
+ any())
+ ).thenReturn(Some(broker1))
+
+ // Build a successful client response.
+ val header = new RequestHeader(ApiKeys.WRITE_TXN_MARKERS, 0, "client", 1)
+ val successfulResponse = new WriteTxnMarkersResponse(
+ Collections.singletonMap(producerId2: java.lang.Long, Collections.singletonMap(partition1, Errors.NONE)))
+ val successfulClientResponse = new ClientResponse(header, null, null,
+ time.milliseconds(), time.milliseconds(), false, null, null,
+ successfulResponse)
+
+ // Build a disconnected client response.
+ val disconnectedClientResponse = new ClientResponse(header, null, null,
+ time.milliseconds(), time.milliseconds(), true, null, null,
+ null)
+
+ // Test matrix to cover various scenarios:
+ val clientResponses = Seq(successfulClientResponse, disconnectedClientResponse)
+ val getTransactionStateResponses = Seq(
+ // NOT_COORDINATOR error case
+ Left(Errors.NOT_COORDINATOR),
+ // COORDINATOR_LOAD_IN_PROGRESS
+ Left(Errors.COORDINATOR_LOAD_IN_PROGRESS),
+ // "Newly loaded" transaction state with the new epoch.
+ Right(Some(CoordinatorEpochAndTxnMetadata(coordinatorEpoch2, txnMetadata2)))
+ )
+
+ clientResponses.foreach { clientResponse =>
+ getTransactionStateResponses.foreach { getTransactionStateResponse =>
+ // Reset data from previous iteration.
+ txnMetadata2.topicPartitions.add(partition1)
+ clearInvocations(txnStateManager)
+ // Send out markers for a transaction before load.
+ channelManager.addTxnMarkersToSend(coordinatorEpoch, txnResult,
+ txnMetadata2, expectedTransition)
+
+ // Drain the marker to make it "in-flight".
+ val requests1 = channelManager.generateRequests().asScala
+ assertEquals(1, requests1.size)
+
+ // Simulate a partition load:
Review Comment:
Ok cool. I'm glad we merged https://github.com/apache/kafka/pull/15139 then. :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15524:
URL: https://github.com/apache/kafka/pull/15524#discussion_r1523937621
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -557,6 +557,7 @@ class TransactionStateManager(brokerId: Int,
loadingPartitions.remove(partitionAndLeaderEpoch)
transactionsPendingForCompletion.foreach { txnTransitMetadata =>
+ info(s"Sending txn markers for $txnTransitMetadata after loading partition $partitionId")
Review Comment:
I appreciate this log as well, but I wonder if we could combine it into a single log message.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #15524:
URL: https://github.com/apache/kafka/pull/15524#discussion_r1526939166
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala:
##########
@@ -177,6 +178,86 @@ class TransactionMarkerChannelManagerTest {
any())
}
+ @Test
+ def shouldNotLoseTxnCompletionAfterLoad(): Unit = {
+ mockCache()
+
+ val expectedTransition = txnMetadata2.prepareComplete(time.milliseconds())
+
+ when(metadataCache.getPartitionLeaderEndpoint(
+ ArgumentMatchers.eq(partition1.topic),
+ ArgumentMatchers.eq(partition1.partition),
+ any())
+ ).thenReturn(Some(broker1))
+
+ // Build a successful client response.
+ val header = new RequestHeader(ApiKeys.WRITE_TXN_MARKERS, 0, "client", 1)
+ val successfulResponse = new WriteTxnMarkersResponse(
+ Collections.singletonMap(producerId2: java.lang.Long, Collections.singletonMap(partition1, Errors.NONE)))
+ val successfulClientResponse = new ClientResponse(header, null, null,
+ time.milliseconds(), time.milliseconds(), false, null, null,
+ successfulResponse)
+
+ // Build a disconnected client response.
+ val disconnectedClientResponse = new ClientResponse(header, null, null,
+ time.milliseconds(), time.milliseconds(), true, null, null,
+ null)
+
+ // Test matrix to cover various scenarios:
+ val clientResponses = Seq(successfulClientResponse, disconnectedClientResponse)
+ val getTransactionStateResponses = Seq(
+ // NOT_COORDINATOR error case
+ Left(Errors.NOT_COORDINATOR),
+ // COORDINATOR_LOAD_IN_PROGRESS
+ Left(Errors.COORDINATOR_LOAD_IN_PROGRESS),
+ // "Newly loaded" transaction state with the new epoch.
+ Right(Some(CoordinatorEpochAndTxnMetadata(coordinatorEpoch2, txnMetadata2)))
+ )
+
+ clientResponses.foreach { clientResponse =>
+ getTransactionStateResponses.foreach { getTransactionStateResponse =>
+ // Reset data from previous iteration.
+ txnMetadata2.topicPartitions.add(partition1)
+ clearInvocations(txnStateManager)
+ // Send out markers for a transaction before load.
+ channelManager.addTxnMarkersToSend(coordinatorEpoch, txnResult,
+ txnMetadata2, expectedTransition)
+
+ // Drain the marker to make it "in-flight".
+ val requests1 = channelManager.generateRequests().asScala
+ assertEquals(1, requests1.size)
+
+ // Simulate a partition load:
Review Comment:
It covers cases in transaction marker request completion handler where we call txnMarkerChannelManager.removeMarkersForTxn (I literally put breakpoints on every line that calls txnMarkerChannelManager.removeMarkersForTxn and ran the test removing breakpoint as it gets hit so at the end of the test there were no breakpoints).
The idea is that previously, there were cases (some more likely, some very unlikely) when a "zombie" reply that was issued before load could unconditionally delete state that was created by newly sent markers that that would lead to transaction being "stuck", now all those cases would leave the state alone so the new transaction completes.
> Is there ever a case of getTransactionStateResponse where we have the old epoch?
in that case we won't try to unconditionally delete txnMarkerChannelManager.removeMarkersForTxn so we won't hit this particular anomaly. The "zombie" reply will complete the transactions, and the reply for the send that happened to be after the load would be a no-op.
We may hit a different anomaly, but assuming that we prevent loading transaction partition with the same coordinator epoch, we should never encounter that case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15524:
URL: https://github.com/apache/kafka/pull/15524#discussion_r1523907297
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala:
##########
@@ -109,23 +109,30 @@ object TransactionMarkerChannelManager {
}
-class TxnMarkerQueue(@volatile var destination: Node) {
+class TxnMarkerQueue(@volatile var destination: Node) extends Logging {
// keep track of the requests per txn topic partition so we can easily clear the queue
// during partition emigration
- private val markersPerTxnTopicPartition = new ConcurrentHashMap[Int, BlockingQueue[TxnIdAndMarkerEntry]]().asScala
+ private val markersPerTxnTopicPartition = new ConcurrentHashMap[Int, BlockingQueue[PendingCompleteTxnAndMarkerEntry]]().asScala
- def removeMarkersForTxnTopicPartition(partition: Int): Option[BlockingQueue[TxnIdAndMarkerEntry]] = {
+ def removeMarkersForTxnTopicPartition(partition: Int): Option[BlockingQueue[PendingCompleteTxnAndMarkerEntry]] = {
markersPerTxnTopicPartition.remove(partition)
}
- def addMarkers(txnTopicPartition: Int, txnIdAndMarker: TxnIdAndMarkerEntry): Unit = {
- val queue = CoreUtils.atomicGetOrUpdate(markersPerTxnTopicPartition, txnTopicPartition,
- new LinkedBlockingQueue[TxnIdAndMarkerEntry]())
- queue.add(txnIdAndMarker)
+ def addMarkers(txnTopicPartition: Int, pendingCompleteTxnAndMarker: PendingCompleteTxnAndMarkerEntry): Unit = {
+ val queue = CoreUtils.atomicGetOrUpdate(markersPerTxnTopicPartition, txnTopicPartition, {
+ info(s"Creating new marker queue for txn partition $txnTopicPartition to destination broker ${destination.id}")
Review Comment:
I noticed in the doc:
```
`createValue` may be invoked more than once if multiple threads attempt to insert a key at the same time
```
This seems ok, but just wanted to call it out
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #15524:
URL: https://github.com/apache/kafka/pull/15524#discussion_r1521968243
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala:
##########
@@ -354,41 +366,42 @@ class TransactionMarkerChannelManager(
txnLogAppend.newMetadata, appendCallback, _ == Errors.COORDINATOR_NOT_AVAILABLE, RequestLocal.NoCaching)
}
- def addTxnMarkersToBrokerQueue(transactionalId: String,
- producerId: Long,
+ def addTxnMarkersToBrokerQueue(producerId: Long,
producerEpoch: Short,
result: TransactionResult,
- coordinatorEpoch: Int,
+ pendingCompleteTxn: PendingCompleteTxn,
Review Comment:
pendingCompleteTxn has transactional id and coordinator epoch, so we don't need to pass them explicitly.
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala:
##########
@@ -109,23 +109,30 @@ object TransactionMarkerChannelManager {
}
-class TxnMarkerQueue(@volatile var destination: Node) {
+class TxnMarkerQueue(@volatile var destination: Node) extends Logging {
// keep track of the requests per txn topic partition so we can easily clear the queue
// during partition emigration
- private val markersPerTxnTopicPartition = new ConcurrentHashMap[Int, BlockingQueue[TxnIdAndMarkerEntry]]().asScala
+ private val markersPerTxnTopicPartition = new ConcurrentHashMap[Int, BlockingQueue[PendingCompleteTxnAndMarkerEntry]]().asScala
Review Comment:
Now we keep track of the PendingCompleteTxn that was added in the transactionsWithPendingMarkers.
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala:
##########
@@ -419,25 +432,34 @@ class TransactionMarkerChannelManager(
def removeMarkersForTxnTopicPartition(txnTopicPartitionId: Int): Unit = {
markersQueueForUnknownBroker.removeMarkersForTxnTopicPartition(txnTopicPartitionId).foreach { queue =>
- for (entry: TxnIdAndMarkerEntry <- queue.asScala)
- removeMarkersForTxnId(entry.txnId)
+ for (entry <- queue.asScala) {
+ info(s"Removing $entry for txn partition $txnTopicPartitionId to destination broker -1")
+ removeMarkersForTxn(entry.pendingCompleteTxn)
+ }
}
- markersQueuePerBroker.foreach { case(_, brokerQueue) =>
+ markersQueuePerBroker.foreach { case(brokerId, brokerQueue) =>
brokerQueue.removeMarkersForTxnTopicPartition(txnTopicPartitionId).foreach { queue =>
- for (entry: TxnIdAndMarkerEntry <- queue.asScala)
- removeMarkersForTxnId(entry.txnId)
+ for (entry <- queue.asScala) {
+ info(s"Removing $entry for txn partition $txnTopicPartitionId to destination broker $brokerId")
+ removeMarkersForTxn(entry.pendingCompleteTxn)
+ }
}
}
}
- def removeMarkersForTxnId(transactionalId: String): Unit = {
- transactionsWithPendingMarkers.remove(transactionalId)
+ def removeMarkersForTxn(pendingCompleteTxn: PendingCompleteTxn): Unit = {
+ val transactionalId = pendingCompleteTxn.transactionalId
+ val removed = transactionsWithPendingMarkers.remove(transactionalId, pendingCompleteTxn)
Review Comment:
This the the core change -- use the original pendingCompleteTxn value to remove the entry. The rest of the change is pretty much plumbing so that we can supply the correct pendingCompleteTxn.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15524:
URL: https://github.com/apache/kafka/pull/15524#discussion_r1523916766
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala:
##########
@@ -419,25 +432,34 @@ class TransactionMarkerChannelManager(
def removeMarkersForTxnTopicPartition(txnTopicPartitionId: Int): Unit = {
markersQueueForUnknownBroker.removeMarkersForTxnTopicPartition(txnTopicPartitionId).foreach { queue =>
- for (entry: TxnIdAndMarkerEntry <- queue.asScala)
- removeMarkersForTxnId(entry.txnId)
+ for (entry <- queue.asScala) {
Review Comment:
Do we have an idea for how many inflight markers we may have at once? Just want to make sure this log message isn't too spammy. I wonder if we could log all the markers in a single log
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15524:
URL: https://github.com/apache/kafka/pull/15524#discussion_r1523732734
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala:
##########
@@ -109,23 +109,30 @@ object TransactionMarkerChannelManager {
}
-class TxnMarkerQueue(@volatile var destination: Node) {
+class TxnMarkerQueue(@volatile var destination: Node) extends Logging {
// keep track of the requests per txn topic partition so we can easily clear the queue
// during partition emigration
- private val markersPerTxnTopicPartition = new ConcurrentHashMap[Int, BlockingQueue[TxnIdAndMarkerEntry]]().asScala
+ private val markersPerTxnTopicPartition = new ConcurrentHashMap[Int, BlockingQueue[PendingCompleteTxnAndMarkerEntry]]().asScala
- def removeMarkersForTxnTopicPartition(partition: Int): Option[BlockingQueue[TxnIdAndMarkerEntry]] = {
+ def removeMarkersForTxnTopicPartition(partition: Int): Option[BlockingQueue[PendingCompleteTxnAndMarkerEntry]] = {
markersPerTxnTopicPartition.remove(partition)
}
- def addMarkers(txnTopicPartition: Int, txnIdAndMarker: TxnIdAndMarkerEntry): Unit = {
- val queue = CoreUtils.atomicGetOrUpdate(markersPerTxnTopicPartition, txnTopicPartition,
- new LinkedBlockingQueue[TxnIdAndMarkerEntry]())
- queue.add(txnIdAndMarker)
+ def addMarkers(txnTopicPartition: Int, pendingCompleteTxnAndMarker: PendingCompleteTxnAndMarkerEntry): Unit = {
+ val queue = CoreUtils.atomicGetOrUpdate(markersPerTxnTopicPartition, txnTopicPartition, {
+ info(s"Creating new marker queue for txn partition $txnTopicPartition to destination broker ${destination.id}")
+ new LinkedBlockingQueue[PendingCompleteTxnAndMarkerEntry]()
+ })
+ queue.add(pendingCompleteTxnAndMarker)
+
+ if (markersPerTxnTopicPartition.get(txnTopicPartition).orNull != queue) {
+ // This could happen if the queue got removed concurrently.
Review Comment:
Should we do something if we added to a "dead queue"?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15524:
URL: https://github.com/apache/kafka/pull/15524#discussion_r1523931941
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerRequestCompletionHandler.scala:
##########
@@ -39,22 +39,23 @@ class TransactionMarkerRequestCompletionHandler(brokerId: Int,
if (response.wasDisconnected) {
trace(s"Cancelled request with header $requestHeader due to node ${response.destination} being disconnected")
- for (txnIdAndMarker <- txnIdAndMarkerEntries.asScala) {
- val transactionalId = txnIdAndMarker.txnId
- val txnMarker = txnIdAndMarker.txnMarkerEntry
+ for (pendingCompleteTxnAndMarker <- pendingCompleteTxnAndMarkerEntries.asScala) {
+ val pendingCompleteTxn = pendingCompleteTxnAndMarker.pendingCompleteTxn
+ val transactionalId = pendingCompleteTxn.transactionalId
+ val txnMarker = pendingCompleteTxnAndMarker.txnMarkerEntry
txnStateManager.getTransactionState(transactionalId) match {
case Left(Errors.NOT_COORDINATOR) =>
info(s"I am no longer the coordinator for $transactionalId; cancel sending transaction markers $txnMarker to the brokers")
- txnMarkerChannelManager.removeMarkersForTxnId(transactionalId)
Review Comment:
unrelated to the changes here, but the first person ("I am") logging here is somewhat strange 😅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #15524:
URL: https://github.com/apache/kafka/pull/15524#issuecomment-2004580064
> LogDirFailureTest > testIOExceptionDuringLogRoll(String) has been failing for a while -- there are some issues I tried to tackle with respect to it, but we ran into some issues. See https://github.com/apache/kafka/pull/15354
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #15524:
URL: https://github.com/apache/kafka/pull/15524#issuecomment-2004976461
I ran "until failure" on trunk and it took about 30 runs to fail.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #15524:
URL: https://github.com/apache/kafka/pull/15524#issuecomment-2004586672
@artemlivshits it may be worth checking if the test is failing on trunk as well. If so we can renew the JIRA to fix it 😅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan merged PR #15524:
URL: https://github.com/apache/kafka/pull/15524
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15524:
URL: https://github.com/apache/kafka/pull/15524#discussion_r1523936892
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerRequestCompletionHandler.scala:
##########
@@ -90,9 +90,10 @@ class TransactionMarkerRequestCompletionHandler(brokerId: Int,
val writeTxnMarkerResponse = response.responseBody.asInstanceOf[WriteTxnMarkersResponse]
val responseErrors = writeTxnMarkerResponse.errorsByProducerId
- for (txnIdAndMarker <- txnIdAndMarkerEntries.asScala) {
Review Comment:
again -- not related to the code changes, but there is a lot of duplicated code between the disconnected and the response cases. Most is the same for the state errors cases like coordinator_load_in_progress etc.
I wonder if we could extract that out and put it above... 🤔
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16352: Txn may get get stuck in PrepareCommit or PrepareAbort s… [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15524:
URL: https://github.com/apache/kafka/pull/15524#discussion_r1525559114
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala:
##########
@@ -177,6 +178,86 @@ class TransactionMarkerChannelManagerTest {
any())
}
+ @Test
+ def shouldNotLoseTxnCompletionAfterLoad(): Unit = {
+ mockCache()
+
+ val expectedTransition = txnMetadata2.prepareComplete(time.milliseconds())
+
+ when(metadataCache.getPartitionLeaderEndpoint(
+ ArgumentMatchers.eq(partition1.topic),
+ ArgumentMatchers.eq(partition1.partition),
+ any())
+ ).thenReturn(Some(broker1))
+
+ // Build a successful client response.
+ val header = new RequestHeader(ApiKeys.WRITE_TXN_MARKERS, 0, "client", 1)
+ val successfulResponse = new WriteTxnMarkersResponse(
+ Collections.singletonMap(producerId2: java.lang.Long, Collections.singletonMap(partition1, Errors.NONE)))
+ val successfulClientResponse = new ClientResponse(header, null, null,
+ time.milliseconds(), time.milliseconds(), false, null, null,
+ successfulResponse)
+
+ // Build a disconnected client response.
+ val disconnectedClientResponse = new ClientResponse(header, null, null,
+ time.milliseconds(), time.milliseconds(), true, null, null,
+ null)
+
+ // Test matrix to cover various scenarios:
+ val clientResponses = Seq(successfulClientResponse, disconnectedClientResponse)
+ val getTransactionStateResponses = Seq(
+ // NOT_COORDINATOR error case
+ Left(Errors.NOT_COORDINATOR),
+ // COORDINATOR_LOAD_IN_PROGRESS
+ Left(Errors.COORDINATOR_LOAD_IN_PROGRESS),
+ // "Newly loaded" transaction state with the new epoch.
+ Right(Some(CoordinatorEpochAndTxnMetadata(coordinatorEpoch2, txnMetadata2)))
+ )
+
+ clientResponses.foreach { clientResponse =>
+ getTransactionStateResponses.foreach { getTransactionStateResponse =>
+ // Reset data from previous iteration.
+ txnMetadata2.topicPartitions.add(partition1)
+ clearInvocations(txnStateManager)
+ // Send out markers for a transaction before load.
+ channelManager.addTxnMarkersToSend(coordinatorEpoch, txnResult,
+ txnMetadata2, expectedTransition)
+
+ // Drain the marker to make it "in-flight".
+ val requests1 = channelManager.generateRequests().asScala
+ assertEquals(1, requests1.size)
+
+ // Simulate a partition load:
Review Comment:
To confirm my understanding, this test simulates sending out marker requests with various responses. We want to send out the requests and then unload and load the pending markers so that we bump the epoch and send out the requests again. This test ensures that when we do these operations, we still correctly complete the transaction when we get a successful response after reloading.
Is there ever a case of `getTransactionStateResponse` where we have the old epoch? I noticed we have the case with the new epoch.
Also in the case with the new epoch, do we want to confirm the transaction doesn't complete?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org