You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jo...@apache.org on 2026/10/13 20:35:43 UTC
(kafka) branch 4.0 updated: KAFKA-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache (4.0) (#23807)
This is an automated email from the ASF dual-hosted git repository.
jolshan pushed a commit to branch 4.0
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.0 by this push:
new 7578266a3fa KAFKA-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache (4.0) (#23807)
7578266a3fa is described below
commit 7578266a3fac7707d9f7c1760686b5aa0bf8864b
Author: Ritika Reddy <98...@users.noreply.github.com>
AuthorDate: Tue Oct 13 13:35:35 2026 -0700
KAFKA-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache (4.0) (#23807)
Backport of #23630 to 4.0 (cherry-pick of
9dfd65db3d5bb3297f29a4a5dea4207658511cd9).
After a broker loses leadership of a `__transaction_state` partition,
its `TransactionStateManager` keeps the partition's cache entry until
`TransactionCoordinator.onResignation` runs, which happens after the
replica has already become a follower. Answers decided purely from that
cache are never fenced, so a TV2 EndTxn(COMMIT) that lands in this
window can get a fatal `INVALID_TXN_STATE` instead of `NOT_COORDINATOR`.
This only uses a cache entry while the local replica still leads the
partition at the entry's coordinator epoch, and returns
`NOT_COORDINATOR` otherwise so the client rediscovers the new
coordinator.
**Backport notes:** the production change is identical to trunk; only
tests were adapted. In addition to the 4.1 adaptations, the new tests
use 4.0's Scala `TransactionState` objects (`Empty`, `Ongoing`,
`CompleteCommit`), its `Pool`-based transaction cache and its
`handleInitProducerId` signature (no 2PC parameters), and
`prepareForTxnMessageAppend` keys the response by `TopicPartition`.
**Testing:** `./gradlew :core:test --tests
'kafka.coordinator.transaction.*' :core:checkstyleTest` passes on 4.0,
including the six new tests.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Reviewers: Justine Olshan <jo...@confluent.io>
Co-authored-by: Claude Opus 5.5 <no...@anthropic.com>
---
.../transaction/TransactionStateManager.scala | 12 +++
.../AbstractCoordinatorConcurrencyTest.scala | 23 ++++-
.../TransactionCoordinatorConcurrencyTest.scala | 102 ++++++++++++++++++++-
.../transaction/TransactionStateManagerTest.scala | 85 +++++++++++++++--
4 files changed, 210 insertions(+), 12 deletions(-)
diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala
index a6e7dd30bf0..2a73c6e6201 100644
--- a/core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala
+++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala
@@ -363,6 +363,14 @@ class TransactionStateManager(brokerId: Int,
}
}
+ /**
+ * Check whether this broker still leads the given transaction state partition at the given coordinator epoch.
+ * The cache entry is only removed on resignation, which happens after the replica becomes a follower.
+ */
+ private def isCoordinatorFor(partitionId: Int, coordinatorEpoch: Int): Boolean =
+ replicaManager.onlinePartition(new TopicPartition(Topic.TRANSACTION_STATE_TOPIC_NAME, partitionId))
+ .exists(partition => partition.isLeader && partition.getLeaderEpoch == coordinatorEpoch)
+
/**
* Get the transaction metadata associated with the given transactional id, or an error if
* the coordinator does not own the transaction partition or is still loading it; if not found
@@ -378,6 +386,10 @@ class TransactionStateManager(brokerId: Int,
Left(Errors.COORDINATOR_LOAD_IN_PROGRESS)
else {
transactionMetadataCache.get(partitionId) match {
+ // leadership moved but the partition has not been unloaded or reloaded yet, so the cache may be stale
+ case Some(cacheEntry) if !isCoordinatorFor(partitionId, cacheEntry.coordinatorEpoch) =>
+ Left(Errors.NOT_COORDINATOR)
+
case Some(cacheEntry) =>
val txnMetadata = Option(cacheEntry.metadataPerTransactionalId.get(transactionalId)).orElse {
createdTxnMetadataOpt.map { createdTxnMetadata =>
diff --git a/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala b/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala
index b574b3dbb0c..119681093a3 100644
--- a/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala
@@ -39,7 +39,9 @@ import org.apache.kafka.server.util.timer.{MockTimer, Timer}
import org.apache.kafka.server.util.{MockScheduler, MockTime, Scheduler}
import org.apache.kafka.storage.internals.log.{AppendOrigin, LogConfig, VerificationGuard}
import org.junit.jupiter.api.{AfterEach, BeforeEach}
-import org.mockito.Mockito.{mock, when, withSettings}
+import org.mockito.Mockito.{RETURNS_DEFAULTS, mock, when, withSettings}
+import org.mockito.invocation.InvocationOnMock
+import org.mockito.stubbing.Answer
import scala.collection._
import scala.jdk.CollectionConverters._
@@ -250,8 +252,25 @@ object AbstractCoordinatorConcurrencyTest {
producePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys.toList.asJava)
}
+ // leader epoch of the partitions this broker leads; None means it is not the leader
+ @volatile var leaderEpochOf: TopicPartition => Option[Int] = _ => None
+ // reuse mocks since the concurrency tests look up partitions frequently
+ private val leaderPartitions = new ConcurrentHashMap[Integer, Partition]()
+
override def onlinePartition(topicPartition: TopicPartition): Option[Partition] = {
- Some(mock(classOf[Partition]))
+ leaderEpochOf(topicPartition) match {
+ case Some(leaderEpoch) =>
+ Some(leaderPartitions.computeIfAbsent(leaderEpoch, _ =>
+ mock(classOf[Partition], withSettings().defaultAnswer(new Answer[AnyRef] {
+ override def answer(invocation: InvocationOnMock): AnyRef = invocation.getMethod.getName match {
+ case "isLeader" => java.lang.Boolean.TRUE
+ case "getLeaderEpoch" => Int.box(leaderEpoch)
+ case _ => RETURNS_DEFAULTS.answer(invocation)
+ }
+ }))))
+ case None =>
+ Some(mock(classOf[Partition]))
+ }
}
def getOrCreateLogs(): mutable.Map[TopicPartition, (UnifiedLog, Long)] = {
diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala
index 70d543bfbe5..76fa670b6c7 100644
--- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala
@@ -43,7 +43,7 @@ import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.mockito.{ArgumentCaptor, ArgumentMatchers}
import org.mockito.ArgumentMatchers.{any, anyInt, anyString}
-import org.mockito.Mockito.{mock, when}
+import org.mockito.Mockito.{mock, never, verify, when}
import scala.jdk.CollectionConverters._
import scala.collection.{Map, mutable}
@@ -69,11 +69,15 @@ class TransactionCoordinatorConcurrencyTest extends AbstractCoordinatorConcurren
val producerId: Long = 11
private var bumpProducerId = false
+ private var pidGenerator: ProducerIdManager = _
@BeforeEach
override def setUp(): Unit = {
super.setUp()
+ // this broker leads all transaction state partitions at coordinatorEpoch
+ replicaManager.leaderEpochOf = tp => if (tp.topic == TRANSACTION_STATE_TOPIC_NAME) Some(coordinatorEpoch) else None
+
val brokerNode = new Node(0, "host", 10)
val metadataCache: MetadataCache = mock(classOf[MetadataCache])
when(metadataCache.getPartitionLeaderEndpoint(
@@ -99,7 +103,7 @@ class TransactionCoordinatorConcurrencyTest extends AbstractCoordinatorConcurren
for (i <- 0 until numPartitions)
txnStateManager.addLoadedTransactionsToCache(i, coordinatorEpoch, new Pool[String, TransactionMetadata]())
- val pidGenerator: ProducerIdManager = mock(classOf[ProducerIdManager])
+ pidGenerator = mock(classOf[ProducerIdManager])
when(pidGenerator.generateProducerId())
.thenAnswer(_ => if (bumpProducerId) {
producerId + 1
@@ -165,6 +169,100 @@ class TransactionCoordinatorConcurrencyTest extends AbstractCoordinatorConcurren
verifyConcurrentRandomSequences(createTransactions, allOperations)
}
+ @Test
+ def testEndTxnReturnsNotCoordinatorAfterLosingLeadershipBeforeResignation(): Unit = {
+ val transactionalId = "stale-coordinator"
+ val producerEpoch: Short = 5
+ txnStateManager.putTransactionStateIfNotExists(new TransactionMetadata(transactionalId, producerId,
+ RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_ID, producerEpoch, (producerEpoch - 1).toShort, 60000,
+ CompleteCommit, mutable.Set.empty[TopicPartition], time.milliseconds(), time.milliseconds(),
+ TransactionVersion.TV_2))
+ val txnPartition = txnStateManager.partitionFor(transactionalId)
+
+ def commit(): Errors = {
+ var error: Errors = null
+ transactionCoordinator.handleEndTransaction(transactionalId, producerId, producerEpoch, TransactionResult.COMMIT,
+ TransactionVersion.TV_2, (e, _, _) => error = e)
+ error
+ }
+
+ // committing at the current epoch with no new transaction is invalid
+ assertEquals(Errors.INVALID_TXN_STATE, commit())
+
+ // the replica became a follower but the partition has not been unloaded yet
+ loseLeadership(txnPartition)
+ assertEquals(Errors.NOT_COORDINATOR, commit())
+ }
+
+ @Test
+ def testVerifyPartitionsReturnsNotCoordinatorAfterLosingLeadershipBeforeResignation(): Unit = {
+ val transactionalId = "stale-coordinator-verify"
+ val producerEpoch: Short = 5
+ val partition = new TopicPartition("topic", 0)
+ txnStateManager.putTransactionStateIfNotExists(new TransactionMetadata(transactionalId, producerId,
+ RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_ID, producerEpoch, (producerEpoch - 1).toShort, 60000,
+ Ongoing, mutable.Set(partition), time.milliseconds(),
+ time.milliseconds(), TransactionVersion.TV_0))
+
+ def verify(): Errors = {
+ var errors: Map[TopicPartition, Errors] = null
+ transactionCoordinator.handleVerifyPartitionsInTransaction(transactionalId, producerId, producerEpoch,
+ Set(partition), result => errors = AddPartitionsToTxnResponse.errorsForTransaction(result.topicResults).asScala)
+ errors(partition)
+ }
+
+ // the partition is part of the ongoing transaction
+ assertEquals(Errors.NONE, verify())
+
+ // the replica became a follower but the partition has not been unloaded yet
+ loseLeadership(txnStateManager.partitionFor(transactionalId))
+ assertEquals(Errors.NOT_COORDINATOR, verify())
+ }
+
+ @Test
+ def testAddPartitionsReturnsNotCoordinatorAfterLosingLeadershipBeforeResignation(): Unit = {
+ val transactionalId = "stale-coordinator-add"
+ val producerEpoch: Short = 5
+ val partition = new TopicPartition("topic", 0)
+ txnStateManager.putTransactionStateIfNotExists(new TransactionMetadata(transactionalId, producerId,
+ RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_ID, producerEpoch, (producerEpoch - 1).toShort, 60000,
+ Ongoing, mutable.Set(partition), time.milliseconds(),
+ time.milliseconds(), TransactionVersion.TV_1))
+
+ def addPartitions(): Errors = {
+ var error: Errors = null
+ transactionCoordinator.handleAddPartitionsToTransaction(transactionalId, producerId, producerEpoch,
+ Set(partition), e => error = e, TransactionVersion.TV_1)
+ error
+ }
+
+ // the partition is already in the ongoing transaction
+ assertEquals(Errors.NONE, addPartitions())
+
+ // the replica became a follower but the partition has not been unloaded yet
+ loseLeadership(txnStateManager.partitionFor(transactionalId))
+ assertEquals(Errors.NOT_COORDINATOR, addPartitions())
+ }
+
+ @Test
+ def testInitProducerIdReturnsNotCoordinatorAfterLosingLeadershipBeforeResignation(): Unit = {
+ val transactionalId = "stale-coordinator-init"
+ loseLeadership(txnStateManager.partitionFor(transactionalId))
+
+ var error: Errors = null
+ transactionCoordinator.handleInitProducerId(transactionalId, 60000, None, result => error = result.error)
+ assertEquals(Errors.NOT_COORDINATOR, error)
+
+ // no producer id is allocated and no entry is added for a partition this broker no longer leads
+ verify(pidGenerator, never()).generateProducerId()
+ replicaManager.leaderEpochOf = tp => if (tp.topic == TRANSACTION_STATE_TOPIC_NAME) Some(coordinatorEpoch) else None
+ assertEquals(Right(None), txnStateManager.getTransactionState(transactionalId))
+ }
+
+ private def loseLeadership(txnPartition: Int): Unit =
+ replicaManager.leaderEpochOf = tp =>
+ if (tp.topic == TRANSACTION_STATE_TOPIC_NAME && tp.partition != txnPartition) Some(coordinatorEpoch) else None
+
/**
* Concurrently load one set of transaction state topic partitions and unload another
* set of partitions. This tests partition leader changes of transaction state topic
diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala
index c3568aa1a88..5e41b2bd1c0 100644
--- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala
@@ -18,9 +18,10 @@ package kafka.coordinator.transaction
import java.lang.management.ManagementFactory
import java.nio.ByteBuffer
-import java.util.concurrent.CountDownLatch
+import java.util.concurrent.{ConcurrentHashMap, CountDownLatch}
import java.util.concurrent.locks.ReentrantLock
import javax.management.ObjectName
+import kafka.cluster.Partition
import kafka.log.UnifiedLog
import kafka.server.{MetadataCache, ReplicaManager}
import kafka.utils.{Pool, TestUtils}
@@ -45,7 +46,9 @@ import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.EnumSource
import org.mockito.{ArgumentCaptor, ArgumentMatchers}
import org.mockito.ArgumentMatchers.{any, anyInt, anyLong, anyShort}
-import org.mockito.Mockito.{atLeastOnce, mock, reset, times, verify, when}
+import org.mockito.Mockito.{RETURNS_DEFAULTS, atLeastOnce, mock, reset, times, verify, when, withSettings}
+import org.mockito.invocation.InvocationOnMock
+import org.mockito.stubbing.Answer
import java.util.Collections
import scala.collection.{Map, mutable}
@@ -63,7 +66,21 @@ class TransactionStateManagerTest {
val time = new MockTime()
val scheduler = new MockScheduler(time)
- val replicaManager: ReplicaManager = mock(classOf[ReplicaManager])
+
+ // (isLeader, leaderEpoch) per transaction state partition; defaults to leading at the loaded epoch
+ var partitionLeadership: Int => (Boolean, Int) = partitionId =>
+ (true, transactionManager.transactionMetadataCache.get(partitionId).map(_.coordinatorEpoch).getOrElse(-1))
+
+ // served by the default answer so that it survives reset(replicaManager)
+ val replicaManager: ReplicaManager = mock(classOf[ReplicaManager], withSettings().defaultAnswer(new Answer[AnyRef] {
+ override def answer(invocation: InvocationOnMock): AnyRef =
+ if (invocation.getMethod.getName == "onlinePartition") {
+ val (isLeader, leaderEpoch) = partitionLeadership(invocation.getArgument[TopicPartition](0).partition)
+ Some(partitionMock(isLeader, leaderEpoch))
+ } else {
+ RETURNS_DEFAULTS.answer(invocation)
+ }
+ }))
val metadataCache: MetadataCache = mock(classOf[MetadataCache])
when(metadataCache.features()).thenReturn {
@@ -126,6 +143,27 @@ class TransactionStateManagerTest {
transactionManager.putTransactionStateIfNotExists(txnMetadata2))
}
+ @Test
+ def testNotCoordinatorAfterLosingLeadershipBeforeResignation(): Unit = {
+ transactionManager.addLoadedTransactionsToCache(partitionId, coordinatorEpoch, new Pool[String, TransactionMetadata]())
+ transactionManager.putTransactionStateIfNotExists(txnMetadata1)
+ assertEquals(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, txnMetadata1))),
+ transactionManager.getTransactionState(transactionalId1))
+
+ // the replica became a follower but the partition has not been unloaded yet
+ partitionLeadership = _ => (false, coordinatorEpoch + 1)
+ assertEquals(Left(Errors.NOT_COORDINATOR), transactionManager.getTransactionState(transactionalId1))
+ assertEquals(Left(Errors.NOT_COORDINATOR), transactionManager.putTransactionStateIfNotExists(txnMetadata2))
+
+ // leader again at a newer epoch but the partition has not been reloaded yet
+ partitionLeadership = _ => (true, coordinatorEpoch + 1)
+ assertEquals(Left(Errors.NOT_COORDINATOR), transactionManager.getTransactionState(transactionalId1))
+
+ // the stale lookup did not add the new transactional id
+ partitionLeadership = _ => (true, coordinatorEpoch)
+ assertEquals(Right(None), transactionManager.getTransactionState(transactionalId2))
+ }
+
@Test
def testDeletePartition(): Unit = {
val metadata1 = transactionMetadata("b", 5L)
@@ -471,6 +509,23 @@ class TransactionStateManagerTest {
transactionManager.appendTransactionToLog(transactionalId1, coordinatorEpoch = 10, failedMetadata, assertCallback, requestLocal = RequestLocal.withThreadConfinedCaching)
}
+ @Test
+ def testNotCoordinatorIfLeadershipLostBeforeAppendCompletes(): Unit = {
+ transactionManager.addLoadedTransactionsToCache(partitionId, coordinatorEpoch, new Pool[String, TransactionMetadata]())
+ transactionManager.putTransactionStateIfNotExists(txnMetadata1)
+
+ // the append succeeds, but the replica becomes a follower before appendTransactionToLog's completion callback runs
+ prepareForTxnMessageAppend(Errors.NONE, () => partitionLeadership = _ => (false, coordinatorEpoch + 1))
+ expectedError = Errors.NOT_COORDINATOR
+ val newMetadata = txnMetadata1.prepareAddPartitions(Set[TopicPartition](new TopicPartition("topic1", 0)), time.milliseconds(), TV_0)
+ transactionManager.appendTransactionToLog(transactionalId1, coordinatorEpoch, newMetadata, assertCallback,
+ requestLocal = RequestLocal.withThreadConfinedCaching)
+
+ // the cache is not updated; the client retries on the new coordinator, which has the record
+ assertEquals(Empty, txnMetadata1.state)
+ assertEquals(Left(Errors.NOT_COORDINATOR), transactionManager.getTransactionState(transactionalId1))
+ }
+
@Test
def testAppendFailToUnknownError(): Unit = {
transactionManager.addLoadedTransactionsToCache(partitionId, coordinatorEpoch, new Pool[String, TransactionMetadata]())
@@ -1193,6 +1248,18 @@ class TransactionStateManagerTest {
assertEquals(expectedError, error)
}
+ private val partitionMocks = new ConcurrentHashMap[(Boolean, Int), Partition]()
+
+ private def partitionMock(isLeader: Boolean, leaderEpoch: Int): Partition =
+ partitionMocks.computeIfAbsent((isLeader, leaderEpoch), _ =>
+ mock(classOf[Partition], withSettings().defaultAnswer(new Answer[AnyRef] {
+ override def answer(invocation: InvocationOnMock): AnyRef = invocation.getMethod.getName match {
+ case "isLeader" => Boolean.box(isLeader)
+ case "getLeaderEpoch" => Int.box(leaderEpoch)
+ case _ => RETURNS_DEFAULTS.answer(invocation)
+ }
+ })))
+
private def transactionMetadata(transactionalId: String,
producerId: Long,
state: TransactionState = Empty,
@@ -1232,7 +1299,7 @@ class TransactionStateManagerTest {
})
}
- private def prepareForTxnMessageAppend(error: Errors): Unit = {
+ private def prepareForTxnMessageAppend(error: Errors, beforeCallback: () => Unit = () => ()): Unit = {
reset(replicaManager)
val capturedArgument: ArgumentCaptor[Map[TopicPartition, PartitionResponse] => Unit] = ArgumentCaptor.forClass(classOf[Map[TopicPartition, PartitionResponse] => Unit])
@@ -1247,10 +1314,12 @@ class TransactionStateManagerTest {
any(),
any(),
any()
- )).thenAnswer(_ => capturedArgument.getValue.apply(
- Map(new TopicPartition(TRANSACTION_STATE_TOPIC_NAME, partitionId) ->
- new PartitionResponse(error, 0L, RecordBatch.NO_TIMESTAMP, 0L)))
- )
+ )).thenAnswer { _ =>
+ beforeCallback()
+ capturedArgument.getValue.apply(
+ Map(new TopicPartition(TRANSACTION_STATE_TOPIC_NAME, partitionId) ->
+ new PartitionResponse(error, 0L, RecordBatch.NO_TIMESTAMP, 0L)))
+ }
}
@Test