[GitHub] [kafka] hachikuji commented on a change in pull request #11725: KAFKA-13221; Implement `PartitionsWithLateTransactionsCount` metric

[GitBox <gi...@apache.org>]

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: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org