hachikuji commented on a change in pull request #11725:
URL: https://github.com/apache/kafka/pull/11725#discussion_r796828017
##########
File path: core/src/main/scala/kafka/log/ProducerStateManager.scala
##########
@@ -498,12 +499,23 @@ class ProducerStateManager(val topicPartition:
TopicPartition,
private var lastMapOffset = 0L
private var lastSnapOffset = 0L
+ // Keep track of the last timestamp from the oldest transaction. This is used
+ // to detect (approximately) when a transaction has been left hanging on a
partition.
+ // We make the field volatile so that it can be safely accessed without a
lock.
+ @volatile private var oldestTxnLastTimestamp: Long = -1L
+
// ongoing transactions sorted by the first offset of the transaction
private val ongoingTxns = new util.TreeMap[Long, TxnMetadata]
// completed transactions whose markers are at offsets above the high
watermark
private val unreplicatedTxns = new util.TreeMap[Long, TxnMetadata]
+ @threadsafe
+ def hasLateTransaction(currentTimeMs: Long): Boolean = {
+ val lastTimestamp = oldestTxnLastTimestamp
+ lastTimestamp > 0 && (currentTimeMs - lastTimestamp) >
maxTransactionTimeoutMs
Review comment:
Yeah, good question. I have been debating it. I had to rely on the last
timestamp here instead of the transaction start time, so that already adds a
slop factor. But is it enough? Perhaps it is better to still add the 5 minutes
to avoid the likelihood of false positives. What do you think?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
