jolshan commented on code in PR #23943:
URL: https://github.com/apache/kafka/pull/23943#discussion_r4383452525
##########
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, (_, _, _, _) => ())
Review Comment:
I was curious about what would happen if we had multiple loads, so glad to
see this 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]
