You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "rreddy-22 (via GitHub)" <gi...@apache.org> on 2026/09/29 23:16:02 UTC
[PR] KAFKA-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache instead of INVALID_TXN_STATE [kafka]
rreddy-22 opened a new pull request, #23630:
URL: https://github.com/apache/kafka/pull/23630
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.
Only use a cache entry while the local replica still leads the partition at the entry's coordinator epoch, and return NOT_COORDINATOR otherwise so the client rediscovers the new coordinato
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4199899386
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
I think that makes sense, I'd still like to keep the current check, since it covers the window after the replica has applied the change but before the cache is unloaded. In that window, any answer served from the cache can be stale (INVALID_TXN_STATE, PRODUCER_FENCED, verification), and this turns them all into NOT_COORDINATOR.
For the metadata-lag case that you brought up, I can implement your suggestion of allowing an empty commit on TV2, handled the same way as empty abort, so a stale coordinator can't complete the write. I think we could do it in a follow up cause we would want to track a state machine change better or we can do it here
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan merged PR #23630:
URL: https://github.com/apache/kafka/pull/23630
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4147163039
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala:
##########
@@ -480,6 +518,23 @@ class TransactionStateManagerTest {
transactionManager.appendTransactionToLog(transactionalId1, coordinatorEpoch = 10, failedMetadata, assertCallback, requestLocal = RequestLocal.withThreadConfinedCaching)
}
+ @Test
+ def testAppendSucceededButLeadershipLostBeforeCallback(): Unit = {
+ transactionManager.addLoadedTransactionsToCache(partitionId, coordinatorEpoch, new ConcurrentHashMap[String, TransactionMetadata]())
+ transactionManager.putTransactionStateIfNotExists(txnMetadata1)
+
+ // the write succeeds but the replica becomes a follower before the callback runs
Review Comment:
I don't quite understand what this test is testing. Which callback is this?
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4147150677
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala:
##########
@@ -172,6 +175,60 @@ 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,
+ TransactionState.COMPLETE_COMMIT, new util.HashSet[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,
+ TransactionState.ONGOING, new util.HashSet[TopicPartition](util.Set.of(partition)), time.milliseconds(),
+ time.milliseconds(), TransactionVersion.TV_0))
+
+ def verify(): Errors = {
+ var errors: util.Map[TopicPartition, Errors] = null
+ transactionCoordinator.handleVerifyPartitionsInTransaction(transactionalId, producerId, producerEpoch,
Review Comment:
did we want to include the other transactional apis (initProducerId, add partitions)?
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4147726878
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala:
##########
@@ -172,6 +175,60 @@ 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,
+ TransactionState.COMPLETE_COMMIT, new util.HashSet[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,
+ TransactionState.ONGOING, new util.HashSet[TopicPartition](util.Set.of(partition)), time.milliseconds(),
+ time.milliseconds(), TransactionVersion.TV_0))
+
+ def verify(): Errors = {
+ var errors: util.Map[TopicPartition, Errors] = null
+ transactionCoordinator.handleVerifyPartitionsInTransaction(transactionalId, producerId, producerEpoch,
Review Comment:
added the tests
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4151983373
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
This doesn't solve the problem if the broker is slow to get updated metadata from the controller. Instead of doing this, we should just allow committing an empty transaction (the same way we allow aborting an empty transaction -- bump the epoch, go through state transitions). If the coordinator is stale, it won't complete because the log won't replicate.
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4185050193
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
I think the problem we were running into is that this was considered an invalid state transition and it was a fatal error to applications like kafka streams. Are you suggesting we should change the state machine to allow empty commits? And are there any adverse implications for that?
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4147613675
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala:
##########
@@ -172,6 +175,60 @@ 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,
+ TransactionState.COMPLETE_COMMIT, new util.HashSet[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,
+ TransactionState.ONGOING, new util.HashSet[TopicPartition](util.Set.of(partition)), time.milliseconds(),
+ time.milliseconds(), TransactionVersion.TV_0))
+
+ def verify(): Errors = {
+ var errors: util.Map[TopicPartition, Errors] = null
+ transactionCoordinator.handleVerifyPartitionsInTransaction(transactionalId, producerId, producerEpoch,
Review Comment:
good point, lemme look into the potential cases there and add tests if needed
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4212631346
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
The check would be the first line of defense, we check if the replica update happened already. If it didn't then we'd go through with the empty commit path. This hopefully reduces the number of times we allow the empty commits and we're still protected from the fatal errors. That's why I'd vote for keeping in the check for sure and adding on the empty commit path for the replica metadata update lag
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4222637538
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
Sure, we can merge this PR in and then change the state machine in a separate PR.
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4149477187
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala:
##########
@@ -480,6 +518,23 @@ class TransactionStateManagerTest {
transactionManager.appendTransactionToLog(transactionalId1, coordinatorEpoch = 10, failedMetadata, assertCallback, requestLocal = RequestLocal.withThreadConfinedCaching)
}
+ @Test
+ def testAppendSucceededButLeadershipLostBeforeCallback(): Unit = {
+ transactionManager.addLoadedTransactionsToCache(partitionId, coordinatorEpoch, new ConcurrentHashMap[String, TransactionMetadata]())
+ transactionManager.putTransactionStateIfNotExists(txnMetadata1)
+
+ // the write succeeds but the replica becomes a follower before the callback runs
Review Comment:
It's the completion callback in appendTransactionToLog (updateCacheCallback), there's a change in behavior cause of our change. If the append succeeds but the replica becomes a follower before that callback runs, the callback's lookup now returns NOT_COORDINATOR instead of NONE. The record is already replicated, so the client's retry on the new coordinator is idempotent. No correctness issues or unexpected behavior
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4201429089
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
I see the trade-offs of having one solution (allow empty commit) could be better than adding multiple places to fail. However, one benefit to your fix is we don't write anything in the log at all. (We have seen some issues when truncation didn't happen properly). I don't feel super strongly either way.
I think my only concern is we are removing a check for a faulty client (ie -- we should never commit an empty transaction from the client perspective) and we are making this change to cover for this race condition -- we don't actually intend to allow empty commits. Still I suppose we will want to do that anyway to cover the slow metadata case and we will handle truncation elsewhere.
TLDR; I can see benefits for doing both fixes, but don't feel strongly about merging both vs just doing the empty commit. If Artem doesn't feel strongly, I can merge this.
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4149583427
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala:
##########
@@ -480,6 +518,23 @@ class TransactionStateManagerTest {
transactionManager.appendTransactionToLog(transactionalId1, coordinatorEpoch = 10, failedMetadata, assertCallback, requestLocal = RequestLocal.withThreadConfinedCaching)
}
+ @Test
+ def testAppendSucceededButLeadershipLostBeforeCallback(): Unit = {
+ transactionManager.addLoadedTransactionsToCache(partitionId, coordinatorEpoch, new ConcurrentHashMap[String, TransactionMetadata]())
+ transactionManager.putTransactionStateIfNotExists(txnMetadata1)
+
+ // the write succeeds but the replica becomes a follower before the callback runs
Review Comment:
Had to think about this a second to make sure there wasn't any potential issue, but makes sense. I guess it can also force the metadata refresh sooner.
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4189028424
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
As far as I can tell allowing committing empty transaction is harmless (and we already do it for aborts). The effect that we'd achieve is that if we're doing it on a stale coordinator (that doesn't know it yet), it will get stuck trying to replicate until the stale coordinator catches up on metadata and then it'll fail with NOT_COORDINATOR error, so the wait for replication naturally covers the race condition window.
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4199899386
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
I think that makes sense, I'd still like to keep the current check, since it covers the window after the replica has applied the change but before the cache is unloaded. In that window, any answer served from the cache can be stale (INVALID_TXN_STATE, PRODUCER_FENCED, verification), and this turns them all into NOT_COORDINATOR.
For the metadata-lag case that you brought up, I can implement your suggestion of allowing an empty commit on TV2, handled the same way as empty abort, so a stale coordinator can't complete the write. I think we should do it in a follow up cause we would want to track a state machine change better
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4201201826
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
What do you think @artemlivshits and @jolshan
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4185954743
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
> This doesn't solve the problem if the broker is slow to get updated metadata from the controller
Yes, this is an issue even during writes though. This race condition returns a stale error even after the replica metadata change takes effect because we return directly from the cache. That would be a slightly different issue but open to discussing it as a follow up once we figure out if there's any side effects, does that sound okay?
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
> This doesn't solve the problem if the broker is slow to get updated metadata from the controller
Yes, this is an issue even during writes though. This race condition returns a stale error even after the replica metadata change takes effect because we return directly from the cache. That would be a slightly different issue but open to discussing it as a follow up once we figure out if there's any side effects, does that sound okay?
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4185050193
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
I think the problem we were running into is that this was considered an invalid state transition and it was a fatal error to applications like kafka streams. Are you suggesting we should change the state machine to allow empty commits?
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4185954743
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
> This doesn't solve the problem if the broker is slow to get updated metadata from the controller
Yes, it doesn't. The race condition we're solving for returns a stale error even after the replica metadata change takes effect because we return directly from the cache.
--
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-21181: Return NOT_COORDINATOR from a stale transaction coordinator cache [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #23630:
URL: https://github.com/apache/kafka/pull/23630#discussion_r4199899386
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -391,6 +391,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))
Review Comment:
I think that makes sense, I'd still like to keep the current check, since it covers the window after the replica has applied the change but before the cache is unloaded. In that window, any answer served from the cache can be stale (INVALID_TXN_STATE, PRODUCER_FENCED, verification), and this turns them all into NOT_COORDINATOR.
For the metadata-lag case that you brought up, I can implement your suggestion of allowing an empty commit on TV2, handled the same way as empty abort, so a stale coordinator can't complete the write. That fixes both the INVALID_TXN_STATE on EndTxn from this ticket. Does that make sense?
--
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