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 2024/11/04 23:32:19 UTC
[PR] KAFKA-15370: Support Participation in 2PC (KIP-939) (1/N) [kafka]
artemlivshits opened a new pull request, #17687:
URL: https://github.com/apache/kafka/pull/17687
This is just a mechanical change to make prepareTransitionTo method use named parameters instead of positional parameters.
--
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-15370: Support Participation in 2PC (KIP-939) (1/N) [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17687:
URL: https://github.com/apache/kafka/pull/17687#discussion_r1893274813
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMetadata.scala:
##########
@@ -329,8 +347,12 @@ private[transaction] class TransactionMetadata(val transactionalId: String,
case _ => txnStartTimestamp
}
- prepareTransitionTo(Ongoing, producerId, producerEpoch, lastProducerEpoch, txnTimeoutMs,
- (topicPartitions ++ addedTopicPartitions).toSet, newTxnStartTimestamp, updateTimestamp, clientTransactionVersion)
+ prepareTransitionTo(
+ state = Ongoing,
+ topicPartitions = (topicPartitions ++ addedTopicPartitions),
+ txnStartTimestamp = newTxnStartTimestamp,
+ txnLastUpdateTimestamp = updateTimestamp
Review Comment:
should we include clientTransactionVersion here since it is now passed to this method too?
--
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-15370: Support Participation in 2PC (KIP-939) (1/N) [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #17687:
URL: https://github.com/apache/kafka/pull/17687#discussion_r1830027618
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMetadata.scala:
##########
@@ -380,50 +417,39 @@ private[transaction] class TransactionMetadata(val transactionalId: String,
}
}
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long): TxnTransitMetadata = {
- prepareTransitionTo(updatedState, updatedProducerId, RecordBatch.NO_PRODUCER_ID, updatedEpoch, updatedLastEpoch, updatedTxnTimeoutMs, updatedTopicPartitions, updatedTxnStartTimestamp, updateTimestamp, TransactionVersion.TV_0)
- }
-
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- nextProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long,
- clientTransactionVersion: TransactionVersion): TxnTransitMetadata = {
+ private def prepareTransitionTo(state: TransactionState,
+ producerId: Long = this.producerId,
+ nextProducerId: Long = this.nextProducerId,
+ producerEpoch: Short = this.producerEpoch,
+ lastProducerEpoch: Short = this.lastProducerEpoch,
+ txnTimeoutMs: Int = this.txnTimeoutMs,
+ topicPartitions: immutable.Set[TopicPartition] = this.topicPartitions.toSet,
+ txnStartTimestamp: Long = this.txnStartTimestamp,
+ txnLastUpdateTimestamp: Long = this.txnLastUpdateTimestamp,
+ clientTransactionVersion: TransactionVersion = this.clientTransactionVersion): TxnTransitMetadata = {
Review Comment:
Adding a summary of offline conversations for visibility:
- we'll keep the existing client version in the state unless it needs to be changed
- we'll update the `completeTransictionTo` to always copy all fields to eliminate a second decision point for which fields are toing to be update; having a second decision point is confusing and can lead to bugs
--
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-15370: Support Participation in 2PC (KIP-939) (1/N) [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17687:
URL: https://github.com/apache/kafka/pull/17687#discussion_r1829726256
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMetadata.scala:
##########
@@ -380,50 +417,39 @@ private[transaction] class TransactionMetadata(val transactionalId: String,
}
}
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long): TxnTransitMetadata = {
- prepareTransitionTo(updatedState, updatedProducerId, RecordBatch.NO_PRODUCER_ID, updatedEpoch, updatedLastEpoch, updatedTxnTimeoutMs, updatedTopicPartitions, updatedTxnStartTimestamp, updateTimestamp, TransactionVersion.TV_0)
- }
-
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- nextProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long,
- clientTransactionVersion: TransactionVersion): TxnTransitMetadata = {
+ private def prepareTransitionTo(state: TransactionState,
+ producerId: Long = this.producerId,
+ nextProducerId: Long = this.nextProducerId,
+ producerEpoch: Short = this.producerEpoch,
+ lastProducerEpoch: Short = this.lastProducerEpoch,
+ txnTimeoutMs: Int = this.txnTimeoutMs,
+ topicPartitions: immutable.Set[TopicPartition] = this.topicPartitions.toSet,
+ txnStartTimestamp: Long = this.txnStartTimestamp,
+ txnLastUpdateTimestamp: Long = this.txnLastUpdateTimestamp,
+ clientTransactionVersion: TransactionVersion = this.clientTransactionVersion): TxnTransitMetadata = {
Review Comment:
Yeah, Things may also change with the change to add partitions implicitly (as we will have the client version there)
--
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-15370: Support Participation in 2PC (KIP-939) (1/N) [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #17687:
URL: https://github.com/apache/kafka/pull/17687#discussion_r1890668553
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMetadataTest.scala:
##########
@@ -231,7 +231,7 @@ class TransactionMetadataTest {
assertEquals(producerId + 1, txnMetadata.producerId)
assertEquals(producerEpoch, txnMetadata.lastProducerEpoch)
assertEquals(0, txnMetadata.producerEpoch)
- assertEquals(1L, txnMetadata.txnStartTimestamp)
+ assertEquals(-1L, txnMetadata.txnStartTimestamp)
Review Comment:
Unconditionally copy data from the transitMetadata when we complete transition.
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMetadataTest.scala:
##########
@@ -204,7 +204,7 @@ class TransactionMetadataTest {
assertEquals(producerId, txnMetadata.producerId)
assertEquals(producerEpoch + 1, txnMetadata.producerEpoch)
assertEquals(producerEpoch, txnMetadata.lastProducerEpoch)
- assertEquals(1L, txnMetadata.txnStartTimestamp)
+ assertEquals(-1L, txnMetadata.txnStartTimestamp)
Review Comment:
This logic is changed (now we unconditionally copy data from the transitMetadata when we complete transition, so original value of 1 is not preserved), so the unit test is changed to codify the new logic. Looking at the history, it doesn't seem like there was a significance in the logic, it's just the UT codified some behavior, and now it codifies a different 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-15370: Support Participation in 2PC (KIP-939) (1/N) [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #17687:
URL: https://github.com/apache/kafka/pull/17687#discussion_r1828556830
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMetadata.scala:
##########
@@ -380,50 +417,39 @@ private[transaction] class TransactionMetadata(val transactionalId: String,
}
}
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long): TxnTransitMetadata = {
- prepareTransitionTo(updatedState, updatedProducerId, RecordBatch.NO_PRODUCER_ID, updatedEpoch, updatedLastEpoch, updatedTxnTimeoutMs, updatedTopicPartitions, updatedTxnStartTimestamp, updateTimestamp, TransactionVersion.TV_0)
- }
-
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- nextProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long,
- clientTransactionVersion: TransactionVersion): TxnTransitMetadata = {
+ private def prepareTransitionTo(state: TransactionState,
+ producerId: Long = this.producerId,
Review Comment:
The defaults just mean that there is no change, so if the caller doesn't use an argument, then it remains unchanged for the transition. I think it makes code easier to read than to require every parameter to be specified: the changed stuff is easy to see from just glancing at the function call, no need to mentally parse which arguments are updated and which stay the same.
--
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-15370: Support Participation in 2PC (KIP-939) (1/N) [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17687:
URL: https://github.com/apache/kafka/pull/17687#discussion_r1828564297
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMetadata.scala:
##########
@@ -380,50 +417,39 @@ private[transaction] class TransactionMetadata(val transactionalId: String,
}
}
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long): TxnTransitMetadata = {
- prepareTransitionTo(updatedState, updatedProducerId, RecordBatch.NO_PRODUCER_ID, updatedEpoch, updatedLastEpoch, updatedTxnTimeoutMs, updatedTopicPartitions, updatedTxnStartTimestamp, updateTimestamp, TransactionVersion.TV_0)
- }
-
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- nextProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long,
- clientTransactionVersion: TransactionVersion): TxnTransitMetadata = {
+ private def prepareTransitionTo(state: TransactionState,
+ producerId: Long = this.producerId,
Review Comment:
I agree with these points. Was just sharing some historical context on this (and other methods). I think it is valid to use 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-15370: Support Participation in 2PC (KIP-939) (1/N) [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17687:
URL: https://github.com/apache/kafka/pull/17687#discussion_r1828546672
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMetadata.scala:
##########
@@ -380,50 +417,39 @@ private[transaction] class TransactionMetadata(val transactionalId: String,
}
}
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long): TxnTransitMetadata = {
- prepareTransitionTo(updatedState, updatedProducerId, RecordBatch.NO_PRODUCER_ID, updatedEpoch, updatedLastEpoch, updatedTxnTimeoutMs, updatedTopicPartitions, updatedTxnStartTimestamp, updateTimestamp, TransactionVersion.TV_0)
- }
-
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- nextProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long,
- clientTransactionVersion: TransactionVersion): TxnTransitMetadata = {
+ private def prepareTransitionTo(state: TransactionState,
+ producerId: Long = this.producerId,
Review Comment:
There was a previous effort to remove default arguments in order to not accidentally set something incorrect. I think the defaults make sense (keeping the same) so I think it makes sense, but wanted to mention this change.
--
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-15370: Support Participation in 2PC (KIP-939) (1/N) [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #17687:
URL: https://github.com/apache/kafka/pull/17687#discussion_r1828614890
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMetadata.scala:
##########
@@ -380,50 +417,39 @@ private[transaction] class TransactionMetadata(val transactionalId: String,
}
}
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long): TxnTransitMetadata = {
- prepareTransitionTo(updatedState, updatedProducerId, RecordBatch.NO_PRODUCER_ID, updatedEpoch, updatedLastEpoch, updatedTxnTimeoutMs, updatedTopicPartitions, updatedTxnStartTimestamp, updateTimestamp, TransactionVersion.TV_0)
- }
-
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- nextProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long,
- clientTransactionVersion: TransactionVersion): TxnTransitMetadata = {
+ private def prepareTransitionTo(state: TransactionState,
+ producerId: Long = this.producerId,
+ nextProducerId: Long = this.nextProducerId,
+ producerEpoch: Short = this.producerEpoch,
+ lastProducerEpoch: Short = this.lastProducerEpoch,
+ txnTimeoutMs: Int = this.txnTimeoutMs,
+ topicPartitions: immutable.Set[TopicPartition] = this.topicPartitions.toSet,
+ txnStartTimestamp: Long = this.txnStartTimestamp,
+ txnLastUpdateTimestamp: Long = this.txnLastUpdateTimestamp,
+ clientTransactionVersion: TransactionVersion = this.clientTransactionVersion): TxnTransitMetadata = {
Review Comment:
Good question. I don't think KIP-939 would rely on persisted value of client version, we should be able to just look at the values of the fields (nextProducerId, nextProducerEpoch) and figure out what logic needs to be done.
But it's a good point that this change actually makes the clientVersion "sticky" in the sense that once it's set it'll keep the value during the transitions. Maybe to keep this change completely mechanical, transition to ongoing or empty should reset the version to TV_0.
--
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-15370: Support Participation in 2PC (KIP-939) (1/N) [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan merged PR #17687:
URL: https://github.com/apache/kafka/pull/17687
--
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-15370: Support Participation in 2PC (KIP-939) (1/N) [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on PR #17687:
URL: https://github.com/apache/kafka/pull/17687#issuecomment-2456017190
Thanks for the PR, makes things much more clear and readable!
--
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-15370: Support Participation in 2PC (KIP-939) (1/N) [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17687:
URL: https://github.com/apache/kafka/pull/17687#discussion_r1828582983
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMetadata.scala:
##########
@@ -380,50 +417,39 @@ private[transaction] class TransactionMetadata(val transactionalId: String,
}
}
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long): TxnTransitMetadata = {
- prepareTransitionTo(updatedState, updatedProducerId, RecordBatch.NO_PRODUCER_ID, updatedEpoch, updatedLastEpoch, updatedTxnTimeoutMs, updatedTopicPartitions, updatedTxnStartTimestamp, updateTimestamp, TransactionVersion.TV_0)
- }
-
- private def prepareTransitionTo(updatedState: TransactionState,
- updatedProducerId: Long,
- nextProducerId: Long,
- updatedEpoch: Short,
- updatedLastEpoch: Short,
- updatedTxnTimeoutMs: Int,
- updatedTopicPartitions: immutable.Set[TopicPartition],
- updatedTxnStartTimestamp: Long,
- updateTimestamp: Long,
- clientTransactionVersion: TransactionVersion): TxnTransitMetadata = {
+ private def prepareTransitionTo(state: TransactionState,
+ producerId: Long = this.producerId,
+ nextProducerId: Long = this.nextProducerId,
+ producerEpoch: Short = this.producerEpoch,
+ lastProducerEpoch: Short = this.lastProducerEpoch,
+ txnTimeoutMs: Int = this.txnTimeoutMs,
+ topicPartitions: immutable.Set[TopicPartition] = this.topicPartitions.toSet,
+ txnStartTimestamp: Long = this.txnStartTimestamp,
+ txnLastUpdateTimestamp: Long = this.txnLastUpdateTimestamp,
+ clientTransactionVersion: TransactionVersion = this.clientTransactionVersion): TxnTransitMetadata = {
Review Comment:
Is the idea to keep clientTransactionVersion the same until each PrepareCommit/PrepareAbortCall? Or will KIP-939 introduce changes to set this in another transactional 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