jolshan commented on code in PR #23943:
URL: https://github.com/apache/kafka/pull/23943#discussion_r4383463815
##########
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:
Any reason to have this comment here but not in the above test?
--
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]
