hachikuji commented on a change in pull request #11725:
URL: https://github.com/apache/kafka/pull/11725#discussion_r797197548
##########
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:
@dajac Thanks, I pushed a commit to add the 5s buffer. It is what the
KIP specifies after all and I couldn't come up with a strong reason not to have
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
