jolshan commented on code in PR #23943:
URL: https://github.com/apache/kafka/pull/23943#discussion_r4383373617
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -605,7 +611,85 @@ class TransactionStateManager(brokerId: Int,
}
val scheduleStartMs = time.milliseconds()
- scheduler.scheduleOnce(s"load-txns-for-partition-$topicPartition", () =>
loadTransactions(scheduleStartMs))
+
+ def scheduleLoad(hwmWaitTimeMs: Long): Unit =
+ scheduler.scheduleOnce(s"load-txns-for-partition-$topicPartition", () =>
loadTransactions(scheduleStartMs, hwmWaitTimeMs))
+
+ // The listener is invoked with partition locks held, so instead of
blocking on stateLock there, this runs on
+ // the scheduler, which also serializes it with the load tasks for the
partition.
+ def scheduleAbandonLoad(reason: String): Unit =
+
scheduler.scheduleOnce(s"abandon-load-txns-for-partition-$topicPartition", ()
=> {
+ inWriteLock[Exception](stateLock, () => {
+ loadingPartitions.remove(partitionAndLeaderEpoch)
+ })
+ info(s"Abandoned loading transaction metadata from $topicPartition at
epoch $coordinatorEpoch because $reason")
+ })
+
+ // The load reads up to the log end offset, but a newly elected leader may
have records above the high watermark
+ // that are not committed yet and may still get truncated. The coordinator
does not write to the partition while
+ // it is loading, so the log end offset doesn't change; wait until the
high watermark catches up to it so that
+ // all the records the load reads are committed.
+ replicaManager.getLog(topicPartition) match {
+ case Some(log) if log.highWatermark < log.logEndOffset =>
+ val logEndOffset = log.logEndOffset
+ info(s"Waiting for the high watermark of $topicPartition (currently
${log.highWatermark}) to reach the log end " +
+ s"offset $logEndOffset before loading transaction metadata at epoch
$coordinatorEpoch")
+ val listener = new HighWatermarkWaitListener(topicPartition,
logEndOffset, coordinatorEpoch, scheduleLoad, scheduleAbandonLoad)
+ if (!replicaManager.maybeAddListener(topicPartition, listener)) {
+ listener.abandon(s"$topicPartition is not online")
+ } else {
+ // the high watermark may have been updated after it was read above
but before the listener was added,
+ // in which case the listener isn't notified of the update: check it
again now that the listener is added
+ listener.onHighWatermarkUpdated(topicPartition, log.highWatermark)
+ // the wait may have finished before the listener was added; in that
case remove it again
+ if (listener.isFinished)
+ replicaManager.removeListener(topicPartition, listener)
+ }
+ case _ =>
+ scheduleLoad(0L)
+ }
+ }
+
+ /**
+ * Waits for the high watermark of a transaction log partition to reach the
given log end offset and then
+ * schedules the load. If this broker stops being the leader first, the load
is abandoned instead.
+ */
+ private class HighWatermarkWaitListener(topicPartition: TopicPartition,
+ logEndOffset: Long,
+ coordinatorEpoch: Int,
+ scheduleLoad: Long => Unit,
+ scheduleAbandonLoad: String => Unit)
extends PartitionListener {
+ private val waitStartMs = time.milliseconds()
+ private val finished = new AtomicBoolean(false)
+
+ def isFinished: Boolean = finished.get
+
+ def abandon(reason: String): Unit = {
+ if (finished.compareAndSet(false, true))
+ finish(scheduleAbandonLoad(reason))
+ }
+
+ private def finish(schedule: => Unit): Unit = {
+ try {
+ schedule
+ } finally {
+ replicaManager.removeListener(topicPartition, this)
+ }
+ }
+
+ override def onHighWatermarkUpdated(partition: TopicPartition, offset:
Long): Unit = {
+ if (offset >= logEndOffset && finished.compareAndSet(false, true)) {
Review Comment:
nice -- we only finish this once per listener. 👍
--
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]
