jolshan commented on code in PR #23943:
URL: https://github.com/apache/kafka/pull/23943#discussion_r4383560553
##########
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:
Looks like we do it after updating follower state, which happens when the
followers fetch. (I was curious if the mechanism was on appending itself or if
it was from fetching)
--
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]
