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 2025/06/25 16:30:41 UTC
[PR] Add initPid Response handling when keepPrepared is set to true [kafka]
rreddy-22 opened a new pull request, #20039:
URL: https://github.com/apache/kafka/pull/20039
When initPid(keepPrepared = true) is called after a client crashes, several situations should be considered.
When there's an ongoing transaction, we can transition it to the newly added PREPARED_TRANSACTION state. However, what if there's no ongoing transaction?
Another scenario could be:
- Issued a commit, to commit prepared
- The commit succeeded on the TC, but the client crashed
- Client restarted with keepPreparedTxn=true (because it doesn't know if the commit succeeded or not and needs to keep retrying the commit until it's successful)
- Issued a commit, but the transaction is not ongoing, because it's committed
**Solution:**
This is a perfectly valid scenario as the external transaction coordinator for the 2PC transaction will keep committing participants, and the participants need to eventually return success (that's a guarantee for a prepared transaction).
_Rejected Alt 1_ -> Return an InvalidTxnStateException : Returning an error would break the above scenario.
_Rejected Alt 2_ -> Then the next thought is that we should somehow validate if the state is expected, but we don't have data to validate the result against.
**Final Solution:** Just returning the success and transitioning to READY is the proper handling of this condition.
--
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-19272: Add initPid Response handling when keepPrepared is set to true (KIP-939) [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #20039:
URL: https://github.com/apache/kafka/pull/20039#discussion_r2211301628
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -1965,6 +1977,9 @@ private PendingStateTransition(
* @return a PreparedTxnState with the current producer ID and epoch
*/
public PreparedTxnState preparedTransactionState() {
- return this.preparedTxnState;
+ if (this.preparedTxnState == ProducerIdAndEpoch.NONE) {
Review Comment:
not sure if it's a big deal either way, but I think it is also ok to not have this check -- as the PreparedTxnState would set the same values.
--
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-19272: Add initPid Response handling when keepPrepared is set to true (KIP-939) [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20039:
URL: https://github.com/apache/kafka/pull/20039#discussion_r2178057965
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -1500,8 +1500,21 @@ public void handleResponse(AbstractResponse response) {
ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(initProducerIdResponse.data().producerId(),
initProducerIdResponse.data().producerEpoch());
setProducerIdAndEpoch(producerIdAndEpoch);
- // TO_DO Add code to handle transition to prepared_txn when keepPrepared = true
- transitionTo(State.READY);
+ // If this is a 2PC-enabled transaction with keepPreparedTxn=true, transition directly
+ // to PREPARED_TRANSACTION state IFF there is an ongoing transaction.
+ if (enable2PC &&
Review Comment:
So irrespective of the enable2PC flag if keepPrepared=true we just transition to prepared state?
--
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-19272: Add initPid Response handling when keepPrepared is set to true (KIP-939) [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20039:
URL: https://github.com/apache/kafka/pull/20039#discussion_r2181020864
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -1500,8 +1500,21 @@ public void handleResponse(AbstractResponse response) {
ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(initProducerIdResponse.data().producerId(),
initProducerIdResponse.data().producerEpoch());
setProducerIdAndEpoch(producerIdAndEpoch);
- // TO_DO Add code to handle transition to prepared_txn when keepPrepared = true
- transitionTo(State.READY);
+ // If this is a 2PC-enabled transaction with keepPreparedTxn=true, transition directly
+ // to PREPARED_TRANSACTION state IFF there is an ongoing transaction.
+ if (enable2PC &&
Review Comment:
Yes, I'm aware that it's allowed, just wanted to confirm the transitions!
--
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-19272: Add initPid Response handling when keepPrepared is set to true (KIP-939) [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #20039:
URL: https://github.com/apache/kafka/pull/20039#discussion_r2176173829
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -1500,8 +1500,21 @@ public void handleResponse(AbstractResponse response) {
ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(initProducerIdResponse.data().producerId(),
initProducerIdResponse.data().producerEpoch());
setProducerIdAndEpoch(producerIdAndEpoch);
- // TO_DO Add code to handle transition to prepared_txn when keepPrepared = true
- transitionTo(State.READY);
+ // If this is a 2PC-enabled transaction with keepPreparedTxn=true, transition directly
+ // to PREPARED_TRANSACTION state IFF there is an ongoing transaction.
+ if (enable2PC &&
Review Comment:
We allow using `keepPreparedTxn` even if `enable2PC == false`.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -1500,8 +1500,21 @@ public void handleResponse(AbstractResponse response) {
ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(initProducerIdResponse.data().producerId(),
initProducerIdResponse.data().producerEpoch());
setProducerIdAndEpoch(producerIdAndEpoch);
- // TO_DO Add code to handle transition to prepared_txn when keepPrepared = true
- transitionTo(State.READY);
+ // If this is a 2PC-enabled transaction with keepPreparedTxn=true, transition directly
+ // to PREPARED_TRANSACTION state IFF there is an ongoing transaction.
+ if (enable2PC &&
+ builder.data.keepPreparedTxn() &&
+ initProducerIdResponse.data().ongoingTxnProducerId() != -1
Review Comment:
There is a `RecordBatch.NO_PRODUCER_ID` constant, we should use it.
--
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-19272: Add initPid Response handling when keepPrepared is set to true (KIP-939) [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #20039:
URL: https://github.com/apache/kafka/pull/20039#discussion_r2178558356
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -1500,8 +1500,20 @@ public void handleResponse(AbstractResponse response) {
ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(initProducerIdResponse.data().producerId(),
initProducerIdResponse.data().producerEpoch());
setProducerIdAndEpoch(producerIdAndEpoch);
- // TO_DO Add code to handle transition to prepared_txn when keepPrepared = true
- transitionTo(State.READY);
+ // If this is a transaction with keepPreparedTxn=true, transition directly
+ // to PREPARED_TRANSACTION state IFF there is an ongoing transaction.
+ if (builder.data.keepPreparedTxn() &&
+ initProducerIdResponse.data().ongoingTxnProducerId() != RecordBatch.NO_PRODUCER_ID
+ ) {
+ transitionTo(State.PREPARED_TRANSACTION);
+ // Update the preparedTxnState with the ongoing pid and epoch from the response.
+ // This will be used to complete the transaction later.
+ String serializedState = initProducerIdResponse.data().ongoingTxnProducerId() +
+ ":" + initProducerIdResponse.data().ongoingTxnProducerEpoch();
+ TransactionManager.this.preparedTxnState = new PreparedTxnState(serializedState);
Review Comment:
Why do we need to serialize (especially custom one-off code) just to get the id / epoch pair again in the state?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -1500,8 +1500,21 @@ public void handleResponse(AbstractResponse response) {
ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(initProducerIdResponse.data().producerId(),
initProducerIdResponse.data().producerEpoch());
setProducerIdAndEpoch(producerIdAndEpoch);
- // TO_DO Add code to handle transition to prepared_txn when keepPrepared = true
- transitionTo(State.READY);
+ // If this is a 2PC-enabled transaction with keepPreparedTxn=true, transition directly
+ // to PREPARED_TRANSACTION state IFF there is an ongoing transaction.
+ if (enable2PC &&
Review Comment:
That's what KIP says: https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=255071659#KIP939:SupportParticipationin2PC-RPCChanges
Could you also check other places in the code that handle `keepPreparedTxn` to make sure we don't accidentally disable if if enable2PC is false?
--
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-19272: Add initPid Response handling when keepPrepared is set to true (KIP-939) [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan merged PR #20039:
URL: https://github.com/apache/kafka/pull/20039
--
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-19272: Add initPid Response handling when keepPrepared is set to true (KIP-939) [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #20039:
URL: https://github.com/apache/kafka/pull/20039#discussion_r2181095646
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -29,6 +29,7 @@
import org.apache.kafka.clients.producer.internals.BufferPool;
import org.apache.kafka.clients.producer.internals.BuiltInPartitioner;
import org.apache.kafka.clients.producer.internals.KafkaProducerMetrics;
+import org.apache.kafka.clients.producer.internals.PreparedTxnState;
Review Comment:
This is public class, it shouldn't be in internals.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -349,7 +348,7 @@ public synchronized void prepareTransaction() {
maybeFailWithError();
transitionTo(State.PREPARED_TRANSACTION);
this.preparedTxnState = new PreparedTxnState(
- this.producerIdAndEpoch.producerId + ":" +
+ this.producerIdAndEpoch.producerId,
Review Comment:
Let's just change the type of `preparedTxnState` member to be `private volatile ProducerIdAndEpoch preparedTxnState` then we can just assign numbers to it (it even has a convenient `NONE` constant). Then the `KafkaProducer` would create the corresponding `PreparedTxnState` object.
--
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-19272: Add initPid Response handling when keepPrepared is set to true (KIP-939) [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20039:
URL: https://github.com/apache/kafka/pull/20039#discussion_r2181020011
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -1500,8 +1500,20 @@ public void handleResponse(AbstractResponse response) {
ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(initProducerIdResponse.data().producerId(),
initProducerIdResponse.data().producerEpoch());
setProducerIdAndEpoch(producerIdAndEpoch);
- // TO_DO Add code to handle transition to prepared_txn when keepPrepared = true
- transitionTo(State.READY);
+ // If this is a transaction with keepPreparedTxn=true, transition directly
+ // to PREPARED_TRANSACTION state IFF there is an ongoing transaction.
+ if (builder.data.keepPreparedTxn() &&
+ initProducerIdResponse.data().ongoingTxnProducerId() != RecordBatch.NO_PRODUCER_ID
+ ) {
+ transitionTo(State.PREPARED_TRANSACTION);
+ // Update the preparedTxnState with the ongoing pid and epoch from the response.
+ // This will be used to complete the transaction later.
+ String serializedState = initProducerIdResponse.data().ongoingTxnProducerId() +
+ ":" + initProducerIdResponse.data().ongoingTxnProducerEpoch();
+ TransactionManager.this.preparedTxnState = new PreparedTxnState(serializedState);
Review Comment:
Just moved the preparedTxnState to the internal directory so I can leverage the other package private constructor
--
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-19272: Add initPid Response handling when keepPrepared is set to true (KIP-939) [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #20039:
URL: https://github.com/apache/kafka/pull/20039#discussion_r2220575422
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -1965,6 +1977,9 @@ private PendingStateTransition(
* @return a PreparedTxnState with the current producer ID and epoch
*/
public PreparedTxnState preparedTransactionState() {
- return this.preparedTxnState;
+ if (this.preparedTxnState == ProducerIdAndEpoch.NONE) {
Review Comment:
The point to use `ProducerIdAndEpoch` in `TransactionManager` is so that we can return it to producer and use the package private constructor to create the object, so that the string representation is encapsulated in the `PreparedTxnState`.
--
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