You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "artemlivshits (via GitHub)" <gi...@apache.org> on 2026/10/23 23:05:24 UTC
[PR] KAFKA-21349: Wait for the high watermark before loading transaction metadata [kafka]
artemlivshits opened a new pull request, #23943:
URL: https://github.com/apache/kafka/pull/23943
When a broker becomes the leader of a `__transaction_state` partition, `TransactionStateManager` loads the log up to the log end offset. A newly elected leader can have records above the high watermark that are not committed yet and can still be truncated if leadership moves again, even with clean elections. The coordinator then acts on state that is later rolled back. For example, it can send COMMIT markers for a PREPARE_COMMIT that the next leader doesn't have, and that leader later times out and aborts the transaction, so the transaction is committed on some partitions and aborted on others.
Loading with `FetchIsolation.HIGH_WATERMARK` is not enough, because a new leader's high watermark can lag behind the true one, so the load would miss committed records.
The coordinator doesn't write to the partition while it is loading (requests get `COORDINATOR_LOAD_IN_PROGRESS`), so the log end offset at the start of the load doesn't change. Once the high watermark reaches it, every record the load reads is committed. This change:
- In `loadTransactionsForTxnTopicPartition`, if the high watermark is below the log end offset, registers a `PartitionListener` and schedules the load once the high watermark reaches the log end offset. The high watermark is checked again after the listener is added, since an update between reading it and adding the listener isn't delivered to the listener.
- Abandons the load (removes the `loadingPartitions` entry) if the partition is not online, becomes a follower, fails or is deleted before the high watermark catches up. This runs on the scheduler, since listeners are invoked with partition locks held.
- Skips partitions that are loading in `removeExpiredTransactionalIds()`. On a leader epoch bump, the previous epoch's cache entry stays until the new load replaces it, and tombstones written for it would move the log end offset while the load waits.
- Logs the time spent waiting for the high watermark as part of the load time.
In the normal case the wait is about one follower fetch round trip. If an ISR follower is unavailable, the load waits until it is removed from the ISR (`replica.lag.time.max.ms`). This does not cover unclean leader election, where committed records can be lost.
Testing: added unit tests to `TransactionStateManagerTest` for waiting for the high watermark, abandoning on each partition transition and when the partition is not online, the high watermark advancing during listener registration, a leader epoch bump while waiting, and expiration of a partition that is being reloaded.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21349: Wait for the high watermark before loading transaction metadata [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan merged PR #23943:
URL: https://github.com/apache/kafka/pull/23943
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21349: Wait for the high watermark before loading transaction metadata [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #23943:
URL: https://github.com/apache/kafka/pull/23943#discussion_r4383325988
##########
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:
What would this case represent?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21349: Wait for the high watermark before loading transaction metadata [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21349: Wait for the high watermark before loading transaction metadata [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #23943:
URL: https://github.com/apache/kafka/pull/23943#discussion_r4383325988
##########
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:
What would this case represent in the case where it is None?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21349: Wait for the high watermark before loading transaction metadata [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21349: Wait for the high watermark before loading transaction metadata [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #23943:
URL: https://github.com/apache/kafka/pull/23943#discussion_r4393659779
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -245,8 +246,12 @@ class TransactionStateManager(brokerId: Int,
private[transaction] def removeExpiredTransactionalIds(): Unit = {
inReadLock[Exception](stateLock, () => {
transactionMetadataCache.foreachEntry { (partitionId, partitionCacheEntry) =>
- val transactionPartition = new TopicPartition(Topic.TRANSACTION_STATE_TOPIC_NAME, partitionId)
- removeExpiredTransactionalIds(transactionPartition, partitionCacheEntry)
+ // the cache entry of a partition that is being reloaded (e.g. after a leader epoch bump) is about to be
+ // replaced, and the load relies on the log end offset not changing while it waits for the high watermark
+ if (!loadingPartitions.exists(_.txnPartitionId == partitionId)) {
Review Comment:
Looks like all other appended records are gated by `getAndMaybeAddTransactionState`. Interesting this one isn't. And good catch :)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21349: Wait for the high watermark before loading transaction metadata [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21349: Wait for the high watermark before loading transaction metadata [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21349: Wait for the high watermark before loading transaction metadata [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #23943:
URL: https://github.com/apache/kafka/pull/23943#discussion_r4383373617
##########
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 _ =>
+ scheduleLoad(0L)
+ }
+ }
+
+ /**
+ * Waits for the high watermark of a transaction log partition to reach the given log end offset and then
+ * schedules the load. If this broker stops being the leader first, the load is abandoned instead.
+ */
+ private class HighWatermarkWaitListener(topicPartition: TopicPartition,
+ logEndOffset: Long,
+ coordinatorEpoch: Int,
+ scheduleLoad: Long => Unit,
+ scheduleAbandonLoad: String => Unit) extends PartitionListener {
+ private val waitStartMs = time.milliseconds()
+ private val finished = new AtomicBoolean(false)
+
+ def isFinished: Boolean = finished.get
+
+ def abandon(reason: String): Unit = {
+ if (finished.compareAndSet(false, true))
+ finish(scheduleAbandonLoad(reason))
+ }
+
+ private def finish(schedule: => Unit): Unit = {
+ try {
+ schedule
+ } finally {
+ replicaManager.removeListener(topicPartition, this)
+ }
+ }
+
+ override def onHighWatermarkUpdated(partition: TopicPartition, offset: Long): Unit = {
+ if (offset >= logEndOffset && finished.compareAndSet(false, true)) {
Review Comment:
nice -- we only finish this once per listener. 👍
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21349: Wait for the high watermark before loading transaction metadata [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org