artemlivshits commented on code in PR #23943:
URL: https://github.com/apache/kafka/pull/23943#discussion_r4391866922
##########
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 _ =>
Review Comment:
If `getLog` returns `None`, the broker has no local log for the partition
(e.g. it was deleted or its log dir went offline between becoming leader and
this call). There is nothing to wait for then: `loadTransactionMetadata`
already handles a missing log by logging a warning and loading nothing, so this
keeps the existing behavior. This branch also covers the high watermark already
being at the log end offset, where everything is committed. I added a comment
to make both cases explicit.
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala:
##########
@@ -201,6 +203,133 @@ class TransactionStateManagerTest {
assertEquals(Some(metadata2.producerEpoch),
cachedProducerEpoch(metadata2.transactionalId))
}
+ @Test
+ def testLoadWaitsForHighWatermarkToReachLogEndOffset(): Unit = {
+ val listener = prepareTxnLogAboveHighWatermark()
+
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch, (_, _, _, _) => ())
+ val partitionListener = listener.getValue
+ assertEquals(Left(Errors.COORDINATOR_LOAD_IN_PROGRESS),
transactionManager.getTransactionState(transactionalId1))
+
+ // the high watermark advances but doesn't reach the log end offset yet
+ partitionListener.onHighWatermarkUpdated(topicPartition, 1L)
+ assertEquals(Left(Errors.COORDINATOR_LOAD_IN_PROGRESS),
transactionManager.getTransactionState(transactionalId1))
+ verify(replicaManager, never()).removeListener(topicPartition,
partitionListener)
+
+ partitionListener.onHighWatermarkUpdated(topicPartition, 2L)
+ verify(replicaManager).removeListener(topicPartition, partitionListener)
+ val cachedMetadata =
transactionManager.getTransactionState(transactionalId1).fold(
+ err => throw new AssertionError(s"$transactionalId1's transaction state
access returns error $err"),
+ entry => entry.getOrElse(throw new AssertionError(s"$transactionalId1's
transaction state was not loaded into the cache"))
+ )
+ assertEquals(txnMetadata1, cachedMetadata.transactionMetadata)
+ assertTrue(transactionManager.loadingPartitions.isEmpty)
+ }
+
+ @Test
+ def testLoadAbandonedWhenPartitionIsNotOnline(): Unit = {
+ prepareTxnLogAboveHighWatermark(online = false)
+
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch, (_, _, _, _) => ())
+ assertEquals(Left(Errors.NOT_COORDINATOR),
transactionManager.getTransactionState(transactionalId1))
+ assertTrue(transactionManager.loadingPartitions.isEmpty)
+ verify(logMockOf(topicPartition), never()).read(anyLong(), anyInt(),
any(), anyBoolean())
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("follower", "failed", "deleted"))
+ def
testLoadAbandonedWhenPartitionTransitionsBeforeHighWatermarkReachesLogEndOffset(transition:
String): Unit = {
+ val listener = prepareTxnLogAboveHighWatermark()
+
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch, (_, _, _, _) => ())
+ val partitionListener = listener.getValue
+ assertEquals(Left(Errors.COORDINATOR_LOAD_IN_PROGRESS),
transactionManager.getTransactionState(transactionalId1))
+
+ transition match {
+ case "follower" => partitionListener.onBecomingFollower(topicPartition)
+ case "failed" => partitionListener.onFailed(topicPartition)
+ case "deleted" => partitionListener.onDeleted(topicPartition)
+ }
+ verify(replicaManager).removeListener(topicPartition, partitionListener)
+ assertEquals(Left(Errors.NOT_COORDINATOR),
transactionManager.getTransactionState(transactionalId1))
+ assertTrue(transactionManager.loadingPartitions.isEmpty)
+ verify(logMockOf(topicPartition), never()).read(anyLong(), anyInt(),
any(), anyBoolean())
+ }
+
+ @Test
+ def testLoadAtNewLeaderEpochWhileWaitingForHighWatermark(): Unit = {
+ val listener = prepareTxnLogAboveHighWatermark()
+
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch, (_, _, _, _) => ())
+ // the leader epoch is bumped while the load at the previous epoch waits
for the high watermark
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch + 1, (_, _, _, _) => ())
+ val Seq(oldEpochListener, newEpochListener) =
listener.getAllValues.asScala.toSeq.takeRight(2)
+ assertEquals(Left(Errors.COORDINATOR_LOAD_IN_PROGRESS),
transactionManager.getTransactionState(transactionalId1))
+
+ oldEpochListener.onHighWatermarkUpdated(topicPartition, 2L)
+ newEpochListener.onHighWatermarkUpdated(topicPartition, 2L)
+ val cachedMetadata =
transactionManager.getTransactionState(transactionalId1).fold(
+ err => throw new AssertionError(s"$transactionalId1's transaction state
access returns error $err"),
+ entry => entry.getOrElse(throw new AssertionError(s"$transactionalId1's
transaction state was not loaded into the cache"))
+ )
+ assertEquals(coordinatorEpoch + 1, cachedMetadata.coordinatorEpoch)
+ assertTrue(transactionManager.loadingPartitions.isEmpty)
+ }
+
+ @Test
+ def
testLoadWhenHighWatermarkReachesLogEndOffsetDuringListenerRegistration(): Unit
= {
+ val listener = prepareTxnLogAboveHighWatermark()
+ val highWatermark = new AtomicLong(0L)
+ when(logMockOf(topicPartition).highWatermark).thenAnswer(_ =>
highWatermark.get)
+ // the high watermark advances after the listener is notified of the
current state but before it's added,
+ // so the listener isn't notified of the update
+ when(replicaManager.maybeAddListener(ArgumentMatchers.eq(topicPartition),
listener.capture())).thenAnswer { _ =>
+ highWatermark.set(2L)
+ true
+ }
+
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch, (_, _, _, _) => ())
+ verify(replicaManager, times(2)).removeListener(topicPartition,
listener.getValue)
+
assertTrue(transactionManager.getTransactionState(transactionalId1).exists(_.isDefined))
+ assertTrue(transactionManager.loadingPartitions.isEmpty)
+ }
+
+ @Test
+ def testLoadWithoutWaitingWhenHighWatermarkAlreadyReachedOnRegistration():
Unit = {
+ val listener = prepareTxnLogAboveHighWatermark()
+ // the high watermark catches up between reading it and registering the
listener
+ when(replicaManager.maybeAddListener(ArgumentMatchers.eq(topicPartition),
listener.capture())).thenAnswer { invocation =>
+
invocation.getArgument[PartitionListener](1).onHighWatermarkUpdated(topicPartition,
2L)
+ true
+ }
+
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch, (_, _, _, _) => ())
+ // removed after registration since the callback ran before the listener
was added
+ verify(replicaManager, times(2)).removeListener(topicPartition,
listener.getValue)
Review Comment:
No good reason: both tests remove the listener twice for similar reasons,
once when the wait finishes and again in
`loadTransactionsForTxnTopicPartition`, because the wait is already finished by
the time registration returns. I added a comment to both tests.
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala:
##########
@@ -201,6 +203,133 @@ class TransactionStateManagerTest {
assertEquals(Some(metadata2.producerEpoch),
cachedProducerEpoch(metadata2.transactionalId))
}
+ @Test
+ def testLoadWaitsForHighWatermarkToReachLogEndOffset(): Unit = {
+ val listener = prepareTxnLogAboveHighWatermark()
+
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch, (_, _, _, _) => ())
+ val partitionListener = listener.getValue
+ assertEquals(Left(Errors.COORDINATOR_LOAD_IN_PROGRESS),
transactionManager.getTransactionState(transactionalId1))
+
+ // the high watermark advances but doesn't reach the log end offset yet
+ partitionListener.onHighWatermarkUpdated(topicPartition, 1L)
+ assertEquals(Left(Errors.COORDINATOR_LOAD_IN_PROGRESS),
transactionManager.getTransactionState(transactionalId1))
+ verify(replicaManager, never()).removeListener(topicPartition,
partitionListener)
+
+ partitionListener.onHighWatermarkUpdated(topicPartition, 2L)
+ verify(replicaManager).removeListener(topicPartition, partitionListener)
+ val cachedMetadata =
transactionManager.getTransactionState(transactionalId1).fold(
+ err => throw new AssertionError(s"$transactionalId1's transaction state
access returns error $err"),
+ entry => entry.getOrElse(throw new AssertionError(s"$transactionalId1's
transaction state was not loaded into the cache"))
+ )
+ assertEquals(txnMetadata1, cachedMetadata.transactionMetadata)
+ assertTrue(transactionManager.loadingPartitions.isEmpty)
+ }
+
+ @Test
+ def testLoadAbandonedWhenPartitionIsNotOnline(): Unit = {
+ prepareTxnLogAboveHighWatermark(online = false)
+
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch, (_, _, _, _) => ())
+ assertEquals(Left(Errors.NOT_COORDINATOR),
transactionManager.getTransactionState(transactionalId1))
+ assertTrue(transactionManager.loadingPartitions.isEmpty)
+ verify(logMockOf(topicPartition), never()).read(anyLong(), anyInt(),
any(), anyBoolean())
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = Array("follower", "failed", "deleted"))
+ def
testLoadAbandonedWhenPartitionTransitionsBeforeHighWatermarkReachesLogEndOffset(transition:
String): Unit = {
+ val listener = prepareTxnLogAboveHighWatermark()
+
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch, (_, _, _, _) => ())
+ val partitionListener = listener.getValue
+ assertEquals(Left(Errors.COORDINATOR_LOAD_IN_PROGRESS),
transactionManager.getTransactionState(transactionalId1))
+
+ transition match {
+ case "follower" => partitionListener.onBecomingFollower(topicPartition)
+ case "failed" => partitionListener.onFailed(topicPartition)
+ case "deleted" => partitionListener.onDeleted(topicPartition)
+ }
+ verify(replicaManager).removeListener(topicPartition, partitionListener)
+ assertEquals(Left(Errors.NOT_COORDINATOR),
transactionManager.getTransactionState(transactionalId1))
+ assertTrue(transactionManager.loadingPartitions.isEmpty)
+ verify(logMockOf(topicPartition), never()).read(anyLong(), anyInt(),
any(), anyBoolean())
+ }
+
+ @Test
+ def testLoadAtNewLeaderEpochWhileWaitingForHighWatermark(): Unit = {
+ val listener = prepareTxnLogAboveHighWatermark()
+
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch, (_, _, _, _) => ())
+ // the leader epoch is bumped while the load at the previous epoch waits
for the high watermark
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch + 1, (_, _, _, _) => ())
+ val Seq(oldEpochListener, newEpochListener) =
listener.getAllValues.asScala.toSeq.takeRight(2)
+ assertEquals(Left(Errors.COORDINATOR_LOAD_IN_PROGRESS),
transactionManager.getTransactionState(transactionalId1))
+
+ oldEpochListener.onHighWatermarkUpdated(topicPartition, 2L)
+ newEpochListener.onHighWatermarkUpdated(topicPartition, 2L)
+ val cachedMetadata =
transactionManager.getTransactionState(transactionalId1).fold(
+ err => throw new AssertionError(s"$transactionalId1's transaction state
access returns error $err"),
+ entry => entry.getOrElse(throw new AssertionError(s"$transactionalId1's
transaction state was not loaded into the cache"))
+ )
+ assertEquals(coordinatorEpoch + 1, cachedMetadata.coordinatorEpoch)
+ assertTrue(transactionManager.loadingPartitions.isEmpty)
+ }
+
+ @Test
+ def
testLoadWhenHighWatermarkReachesLogEndOffsetDuringListenerRegistration(): Unit
= {
+ val listener = prepareTxnLogAboveHighWatermark()
+ val highWatermark = new AtomicLong(0L)
+ when(logMockOf(topicPartition).highWatermark).thenAnswer(_ =>
highWatermark.get)
+ // the high watermark advances after the listener is notified of the
current state but before it's added,
+ // so the listener isn't notified of the update
+ when(replicaManager.maybeAddListener(ArgumentMatchers.eq(topicPartition),
listener.capture())).thenAnswer { _ =>
+ highWatermark.set(2L)
+ true
+ }
+
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch, (_, _, _, _) => ())
+ verify(replicaManager, times(2)).removeListener(topicPartition,
listener.getValue)
+
assertTrue(transactionManager.getTransactionState(transactionalId1).exists(_.isDefined))
+ assertTrue(transactionManager.loadingPartitions.isEmpty)
+ }
+
+ @Test
+ def testLoadWithoutWaitingWhenHighWatermarkAlreadyReachedOnRegistration():
Unit = {
+ val listener = prepareTxnLogAboveHighWatermark()
+ // the high watermark catches up between reading it and registering the
listener
+ when(replicaManager.maybeAddListener(ArgumentMatchers.eq(topicPartition),
listener.capture())).thenAnswer { invocation =>
+
invocation.getArgument[PartitionListener](1).onHighWatermarkUpdated(topicPartition,
2L)
+ true
+ }
+
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch, (_, _, _, _) => ())
+ // removed after registration since the callback ran before the listener
was added
+ verify(replicaManager, times(2)).removeListener(topicPartition,
listener.getValue)
+
assertTrue(transactionManager.getTransactionState(transactionalId1).exists(_.isDefined))
+ }
+
+ // Prepares a log with two records for transactionalId1 where the high
watermark is below the log end offset
+ private def prepareTxnLogAboveHighWatermark(online: Boolean = true):
ArgumentCaptor[PartitionListener] = {
Review Comment:
Yes, it is driven by the followers fetching: the leader's high watermark
advances in `maybeIncrementLeaderHW` after `updateFollowerFetchState`, once all
ISR members have fetched up to an offset. It can also advance when a lagging
follower is removed from the ISR (`maybeShrinkIsr`), on `makeLeader`, and on
appends to the leader. The coordinator does not append while loading, so in
practice it is the follower fetches, typically one fetch round trip after the
election. I added a comment to the test helper saying that the tests drive the
listener directly while on a real leader the updates come from follower
fetches, and a line in the main code saying where the high watermark advances.
--
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]
