jolshan commented on code in PR #23943:
URL: https://github.com/apache/kafka/pull/23943#discussion_r4383532340
##########
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:
One thing I wasn't totally sure about was in the scenario we saw, was when
the high watermark updated on the leader.
Ie, in these testing scenarios, we are manually setting the high watermark
to be updated, but how does it work in reality? Let me read
https://github.com/apache/kafka/blob/2bd1d20063ce7e0843bb1482f7252b67c2a40421/core/src/main/scala/kafka/cluster/Partition.scala#L1013
again. 😂
--
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]
