You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "jolshan (via GitHub)" <gi...@apache.org> on 2023/12/28 23:53:19 UTC
[PR] KAFKA-15987: Refactor ReplicaManager code for transaction verification [kafka]
jolshan opened a new pull request, #15087:
URL: https://github.com/apache/kafka/pull/15087
I originally did some refactors in https://github.com/apache/kafka/pull/14774, but we decided to keep the changes minimal since the ticket was a blocker. Here are those refactors:
* Removed separate append paths so that produce, group coordinator, and other append paths all call appendRecords
* appendRecords has been simplified
* Removed unneeded error conversions in verification code since group coordinator and produce path convert errors differently, removed test for that
* Fixed incorrect capital param name in KafkaRequestHandler
* Updated ReplicaManager test to handle transaction verification appends separately. (I may revise this)
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan merged PR #15087:
URL: https://github.com/apache/kafka/pull/15087
--
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-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15087:
URL: https://github.com/apache/kafka/pull/15087#discussion_r1458144631
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2560,41 +2562,6 @@ class ReplicaManagerTest {
}
}
- @ParameterizedTest
- @EnumSource(value = classOf[Errors], names = Array("NOT_COORDINATOR", "CONCURRENT_TRANSACTIONS", "COORDINATOR_LOAD_IN_PROGRESS", "COORDINATOR_NOT_AVAILABLE"))
- def testMaybeVerificationErrorConversions(error: Errors): Unit = {
Review Comment:
We also have one for the GroupCoordinator.
--
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-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15087:
URL: https://github.com/apache/kafka/pull/15087#discussion_r1458144452
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2560,41 +2562,6 @@ class ReplicaManagerTest {
}
}
- @ParameterizedTest
- @EnumSource(value = classOf[Errors], names = Array("NOT_COORDINATOR", "CONCURRENT_TRANSACTIONS", "COORDINATOR_LOAD_IN_PROGRESS", "COORDINATOR_NOT_AVAILABLE"))
- def testMaybeVerificationErrorConversions(error: Errors): Unit = {
Review Comment:
We have the test above in this file.
--
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-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "hachikuji (via GitHub)" <gi...@apache.org>.
hachikuji commented on code in PR #15087:
URL: https://github.com/apache/kafka/pull/15087#discussion_r1441051073
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -749,9 +749,11 @@ class ReplicaManager(val config: KafkaConfig,
* @param responseCallback callback for sending the response
* @param delayedProduceLock lock for the delayed actions
* @param recordValidationStatsCallback callback for updating stats on record conversions
- * @param requestLocal container for the stateful instances scoped to this request
- * @param transactionalId transactional ID if the request is from a producer and the producer is transactional
+ * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
+ * thread calling this method
* @param actionQueue the action queue to use. ReplicaManager#defaultActionQueue is used by default.
+ * @param verificationGuards the mapping from topic partition to verification guards if transaction verification is used
+ * @param preAppendErrors the mapping from topic partition to LogAppendResult for errors that occurred before appending
Review Comment:
This seems a little strange. If the partitions have already failed, then why do we need to pass them through `appendRecords`? I would expect instead that the caller would just join the pre-append failures with the append failures.
--
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-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15087:
URL: https://github.com/apache/kafka/pull/15087#discussion_r1441068240
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -749,9 +749,11 @@ class ReplicaManager(val config: KafkaConfig,
* @param responseCallback callback for sending the response
* @param delayedProduceLock lock for the delayed actions
* @param recordValidationStatsCallback callback for updating stats on record conversions
- * @param requestLocal container for the stateful instances scoped to this request
- * @param transactionalId transactional ID if the request is from a producer and the producer is transactional
+ * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
+ * thread calling this method
* @param actionQueue the action queue to use. ReplicaManager#defaultActionQueue is used by default.
+ * @param verificationGuards the mapping from topic partition to verification guards if transaction verification is used
+ * @param preAppendErrors the mapping from topic partition to LogAppendResult for errors that occurred before appending
Review Comment:
Oh I see what you are saying -- basically the appendCallback should be defined so the verification errors are joined whenever the callback is defined (ideally after we get the verification errors).
So the callback for verification would include the definition of the append callback. I think this could work.
Time to change all the mock code 😂
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15087:
URL: https://github.com/apache/kafka/pull/15087#discussion_r1458066834
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -717,17 +717,16 @@ class KafkaApis(val requestChannel: RequestChannel,
val internalTopicsAllowed = request.header.clientId == AdminUtils.ADMIN_CLIENT_ID
// call the replica manager to append messages to the replicas
- replicaManager.appendRecords(
+ replicaManager.handleProduceAppend(
timeout = produceRequest.timeout.toLong,
requiredAcks = produceRequest.acks,
internalTopicsAllowed = internalTopicsAllowed,
origin = AppendOrigin.CLIENT,
Review Comment:
that's fair. I can do 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-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #15087:
URL: https://github.com/apache/kafka/pull/15087#discussion_r2296789472
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -762,167 +763,123 @@ class ReplicaManager(val config: KafkaConfig,
delayedProduceLock: Option[Lock] = None,
recordValidationStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit = _ => (),
requestLocal: RequestLocal = RequestLocal.NoCaching,
- transactionalId: String = null,
- actionQueue: ActionQueue = this.defaultActionQueue): Unit = {
- if (isValidRequiredAcks(requiredAcks)) {
-
- val verificationGuards: mutable.Map[TopicPartition, VerificationGuard] = mutable.Map[TopicPartition, VerificationGuard]()
- val (verifiedEntriesPerPartition, notYetVerifiedEntriesPerPartition, errorsPerPartition) =
- if (transactionalId == null || !config.transactionPartitionVerificationEnable)
- (entriesPerPartition, Map.empty[TopicPartition, MemoryRecords], Map.empty[TopicPartition, Errors])
- else {
- val verifiedEntries = mutable.Map[TopicPartition, MemoryRecords]()
- val unverifiedEntries = mutable.Map[TopicPartition, MemoryRecords]()
- val errorEntries = mutable.Map[TopicPartition, Errors]()
- partitionEntriesForVerification(verificationGuards, entriesPerPartition, verifiedEntries, unverifiedEntries, errorEntries)
- (verifiedEntries.toMap, unverifiedEntries.toMap, errorEntries.toMap)
- }
-
- if (notYetVerifiedEntriesPerPartition.isEmpty || addPartitionsToTxnManager.isEmpty) {
- appendEntries(verifiedEntriesPerPartition, internalTopicsAllowed, origin, requiredAcks, verificationGuards.toMap,
- errorsPerPartition, recordValidationStatsCallback, timeout, responseCallback, delayedProduceLock, actionQueue)(requestLocal, Map.empty)
- } else {
- // For unverified entries, send a request to verify. When verified, the append process will proceed via the callback.
- // We verify above that all partitions use the same producer ID.
- val batchInfo = notYetVerifiedEntriesPerPartition.head._2.firstBatch()
- addPartitionsToTxnManager.foreach(_.verifyTransaction(
- transactionalId = transactionalId,
- producerId = batchInfo.producerId,
- producerEpoch = batchInfo.producerEpoch,
- topicPartitions = notYetVerifiedEntriesPerPartition.keySet.toSeq,
- callback = KafkaRequestHandler.wrapAsyncCallback(
- appendEntries(
- entriesPerPartition,
- internalTopicsAllowed,
- origin,
- requiredAcks,
- verificationGuards.toMap,
- errorsPerPartition,
- recordValidationStatsCallback,
- timeout,
- responseCallback,
- delayedProduceLock,
- actionQueue
- ),
- requestLocal)
- ))
- }
- } else {
- // If required.acks is outside accepted range, something is wrong with the client
- // Just return an error and don't handle the request at all
- val responseStatus = entriesPerPartition.map { case (topicPartition, _) =>
- topicPartition -> new PartitionResponse(
- Errors.INVALID_REQUIRED_ACKS,
- LogAppendInfo.UNKNOWN_LOG_APPEND_INFO.firstOffset,
- RecordBatch.NO_TIMESTAMP,
- LogAppendInfo.UNKNOWN_LOG_APPEND_INFO.logStartOffset
- )
- }
- responseCallback(responseStatus)
+ actionQueue: ActionQueue = this.defaultActionQueue,
+ verificationGuards: Map[TopicPartition, VerificationGuard] = Map.empty): Unit = {
+ if (!isValidRequiredAcks(requiredAcks)) {
+ sendInvalidRequiredAcksResponse(entriesPerPartition, responseCallback)
+ return
}
- }
- /*
- * Note: This method can be used as a callback in a different request thread. Ensure that correct RequestLocal
- * is passed when executing this method. Accessing non-thread-safe data structures should be avoided if possible.
- */
- private def appendEntries(allEntries: Map[TopicPartition, MemoryRecords],
- internalTopicsAllowed: Boolean,
- origin: AppendOrigin,
- requiredAcks: Short,
- verificationGuards: Map[TopicPartition, VerificationGuard],
- errorsPerPartition: Map[TopicPartition, Errors],
- recordConversionStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit,
- timeout: Long,
- responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
- delayedProduceLock: Option[Lock],
- actionQueue: ActionQueue)
- (requestLocal: RequestLocal, unverifiedEntries: Map[TopicPartition, Errors]): Unit = {
val sTime = time.milliseconds
- val verifiedEntries =
- if (unverifiedEntries.isEmpty)
- allEntries
- else
- allEntries.filter { case (tp, _) =>
- !unverifiedEntries.contains(tp)
- }
-
val localProduceResults = appendToLocalLog(internalTopicsAllowed = internalTopicsAllowed,
- origin, verifiedEntries, requiredAcks, requestLocal, verificationGuards.toMap)
+ origin, entriesPerPartition, requiredAcks, requestLocal, verificationGuards.toMap)
debug("Produce to local log in %d ms".format(time.milliseconds - sTime))
- val errorResults = (unverifiedEntries ++ errorsPerPartition).map {
- case (topicPartition, error) =>
- // translate transaction coordinator errors to known producer response errors
- val customException =
- error match {
- case Errors.INVALID_TXN_STATE => Some(error.exception("Partition was not added to the transaction"))
- case Errors.CONCURRENT_TRANSACTIONS |
- Errors.COORDINATOR_LOAD_IN_PROGRESS |
- Errors.COORDINATOR_NOT_AVAILABLE |
- Errors.NOT_COORDINATOR => Some(new NotEnoughReplicasException(
- s"Unable to verify the partition has been added to the transaction. Underlying error: ${error.toString}"))
- case _ => None
- }
- topicPartition -> LogAppendResult(
- LogAppendInfo.UNKNOWN_LOG_APPEND_INFO,
- Some(customException.getOrElse(error.exception)),
- hasCustomErrorMessage = customException.isDefined
- )
- }
-
- val allResults = localProduceResults ++ errorResults
- val produceStatus = buildProducePartitionStatus(allResults)
+ val produceStatus = buildProducePartitionStatus(localProduceResults)
- addCompletePurgatoryAction(actionQueue, allResults)
- recordConversionStatsCallback(localProduceResults.map { case (k, v) =>
+ addCompletePurgatoryAction(actionQueue, localProduceResults)
+ recordValidationStatsCallback(localProduceResults.map { case (k, v) =>
k -> v.info.recordValidationStats
})
maybeAddDelayedProduce(
requiredAcks,
delayedProduceLock,
timeout,
- allEntries,
- allResults,
+ entriesPerPartition,
+ localProduceResults,
produceStatus,
responseCallback
)
}
- private def partitionEntriesForVerification(verificationGuards: mutable.Map[TopicPartition, VerificationGuard],
- entriesPerPartition: Map[TopicPartition, MemoryRecords],
- verifiedEntries: mutable.Map[TopicPartition, MemoryRecords],
- unverifiedEntries: mutable.Map[TopicPartition, MemoryRecords],
- errorEntries: mutable.Map[TopicPartition, Errors]): Unit = {
- val transactionalProducerIds = mutable.HashSet[Long]()
+ /**
+ * Handles the produce request by starting any transactional verification before appending.
+ *
+ * @param timeout maximum time we will wait to append before returning
+ * @param requiredAcks number of replicas who must acknowledge the append before sending the response
+ * @param internalTopicsAllowed boolean indicating whether internal topics can be appended to
+ * @param transactionalId the transactional ID for the produce request or null if there is none.
+ * @param entriesPerPartition the records per partition to be appended
+ * @param responseCallback callback for sending the response
+ * @param recordValidationStatsCallback callback for updating stats on record conversions
+ * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
+ * thread calling this method
+ * @param actionQueue the action queue to use. ReplicaManager#defaultActionQueue is used by default.
+ */
+ def handleProduceAppend(timeout: Long,
+ requiredAcks: Short,
+ internalTopicsAllowed: Boolean,
+ transactionalId: String,
+ entriesPerPartition: Map[TopicPartition, MemoryRecords],
+ responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
+ recordValidationStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit = _ => (),
+ requestLocal: RequestLocal = RequestLocal.NoCaching,
+ actionQueue: ActionQueue = this.defaultActionQueue): Unit = {
+
+ val transactionalProducerInfo = mutable.HashSet[(Long, Short)]()
+ val topicPartitionBatchInfo = mutable.Map[TopicPartition, Int]()
entriesPerPartition.foreach { case (topicPartition, records) =>
- try {
- // Produce requests (only requests that require verification) should only have one batch per partition in "batches" but check all just to be safe.
- val transactionalBatches = records.batches.asScala.filter(batch => batch.hasProducerId && batch.isTransactional)
- transactionalBatches.foreach(batch => transactionalProducerIds.add(batch.producerId))
-
- if (transactionalBatches.nonEmpty) {
- // We return VerificationGuard if the partition needs to be verified. If no state is present, no need to verify.
- val firstBatch = records.firstBatch
- val verificationGuard = getPartitionOrException(topicPartition).maybeStartTransactionVerification(firstBatch.producerId, firstBatch.baseSequence, firstBatch.producerEpoch)
- if (verificationGuard != VerificationGuard.SENTINEL) {
- verificationGuards.put(topicPartition, verificationGuard)
- unverifiedEntries.put(topicPartition, records)
- } else
- verifiedEntries.put(topicPartition, records)
- } else {
- // If there is no producer ID or transactional records in the batches, no need to verify.
- verifiedEntries.put(topicPartition, records)
- }
- } catch {
- case e: Exception => errorEntries.put(topicPartition, Errors.forException(e))
- }
+ // Produce requests (only requests that require verification) should only have one batch per partition in "batches" but check all just to be safe.
+ val transactionalBatches = records.batches.asScala.filter(batch => batch.hasProducerId && batch.isTransactional)
+ transactionalBatches.foreach(batch => transactionalProducerInfo.add(batch.producerId, batch.producerEpoch))
+ if (!transactionalBatches.isEmpty) topicPartitionBatchInfo.put(topicPartition, records.firstBatch.baseSequence)
}
- // We should have exactly one producer ID for transactional records
- if (transactionalProducerIds.size > 1) {
+ if (transactionalProducerInfo.size > 1) {
throw new InvalidPidMappingException("Transactional records contained more than one producer ID")
}
+
+ def postVerificationCallback(preAppendErrors: Map[TopicPartition, Errors],
+ newRequestLocal: RequestLocal,
+ verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
+ val errorResults = preAppendErrors.map {
+ case (topicPartition, error) =>
+ // translate transaction coordinator errors to known producer response errors
+ val customException =
+ error match {
+ case Errors.INVALID_TXN_STATE => Some(error.exception("Partition was not added to the transaction"))
+ case Errors.CONCURRENT_TRANSACTIONS |
+ Errors.COORDINATOR_LOAD_IN_PROGRESS |
+ Errors.COORDINATOR_NOT_AVAILABLE |
+ Errors.NOT_COORDINATOR => Some(new NotEnoughReplicasException(
+ s"Unable to verify the partition has been added to the transaction. Underlying error: ${error.toString}"))
+ case _ => None
+ }
+ topicPartition -> LogAppendResult(
+ LogAppendInfo.UNKNOWN_LOG_APPEND_INFO,
+ Some(customException.getOrElse(error.exception)),
+ hasCustomErrorMessage = customException.isDefined
+ )
+ }
+ val entriesWithoutErrorsPerPartition = entriesPerPartition.filter { case (key, _) => !errorResults.contains(key) }
Review Comment:
Sorry for raising a question on this merged PR. I just have one small concern: in this non-transaction path, `errorResults` is always empty, so there is no need to recreate a map collection, since `entriesWithoutErrorsPerPartition` is identical to `entriesPerPartition`. Did I misunderstand anything? If not, I can file a minor PR to improve 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-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15087:
URL: https://github.com/apache/kafka/pull/15087#discussion_r1458143630
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1188,18 +1073,7 @@ class ReplicaManager(val config: KafkaConfig,
requestLocal: RequestLocal,
verificationErrors: Map[TopicPartition, Errors]
): Unit = {
- // Map transaction coordinator errors to known errors for the response
- val convertedErrors = verificationErrors.map { case (tp, error) =>
- error match {
- case Errors.CONCURRENT_TRANSACTIONS |
- Errors.COORDINATOR_LOAD_IN_PROGRESS |
- Errors.COORDINATOR_NOT_AVAILABLE |
- Errors.NOT_COORDINATOR => tp -> Errors.NOT_ENOUGH_REPLICAS
- case _ => tp -> error
- }
-
- }
Review Comment:
We have separate handling for produce requests and txn offset commit requests.
for produce:
```
case Errors.INVALID_TXN_STATE => Some(error.exception("Partition was not added to the transaction"))
case Errors.CONCURRENT_TRANSACTIONS |
Errors.COORDINATOR_LOAD_IN_PROGRESS |
Errors.COORDINATOR_NOT_AVAILABLE |
Errors.NOT_COORDINATOR => Some(new NotEnoughReplicasException(
s"Unable to verify the partition has been added to the transaction. Underlying error: ${error.toString}"))
case _ => None
```
for txn offset commit:
```
error match {
case Errors.UNKNOWN_TOPIC_OR_PARTITION
| Errors.NOT_ENOUGH_REPLICAS
| Errors.NOT_ENOUGH_REPLICAS_AFTER_APPEND =>
Errors.COORDINATOR_NOT_AVAILABLE
case Errors.NOT_LEADER_OR_FOLLOWER
| Errors.KAFKA_STORAGE_ERROR =>
Errors.NOT_COORDINATOR
case Errors.MESSAGE_TOO_LARGE
| Errors.RECORD_LIST_TOO_LARGE
| Errors.INVALID_FETCH_SIZE =>
Errors.INVALID_COMMIT_OFFSET_SIZE
// We may see INVALID_TXN_STATE or INVALID_PID_MAPPING here due to transaction verification.
// They can be returned without mapping to a new error.
case other => other
}
```
--
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-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15087:
URL: https://github.com/apache/kafka/pull/15087#discussion_r1458143981
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1188,18 +1073,7 @@ class ReplicaManager(val config: KafkaConfig,
requestLocal: RequestLocal,
verificationErrors: Map[TopicPartition, Errors]
): Unit = {
- // Map transaction coordinator errors to known errors for the response
- val convertedErrors = verificationErrors.map { case (tp, error) =>
- error match {
- case Errors.CONCURRENT_TRANSACTIONS |
- Errors.COORDINATOR_LOAD_IN_PROGRESS |
- Errors.COORDINATOR_NOT_AVAILABLE |
- Errors.NOT_COORDINATOR => tp -> Errors.NOT_ENOUGH_REPLICAS
- case _ => tp -> error
- }
-
- }
Review Comment:
But yes, we simply pass through concurrent txns which will be fatal to the client.
--
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-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15087:
URL: https://github.com/apache/kafka/pull/15087#discussion_r1441066328
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -749,9 +749,11 @@ class ReplicaManager(val config: KafkaConfig,
* @param responseCallback callback for sending the response
* @param delayedProduceLock lock for the delayed actions
* @param recordValidationStatsCallback callback for updating stats on record conversions
- * @param requestLocal container for the stateful instances scoped to this request
- * @param transactionalId transactional ID if the request is from a producer and the producer is transactional
+ * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
+ * thread calling this method
* @param actionQueue the action queue to use. ReplicaManager#defaultActionQueue is used by default.
+ * @param verificationGuards the mapping from topic partition to verification guards if transaction verification is used
+ * @param preAppendErrors the mapping from topic partition to LogAppendResult for errors that occurred before appending
Review Comment:
I was considering a write where some partitions were verified and some were not. We would allow the ones that succeeded to still write and not the ones that failed verification.
Not sure if we support multi-partition writes in the produce api, but it appears we do.
--
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-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15087:
URL: https://github.com/apache/kafka/pull/15087#discussion_r1443537258
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -749,9 +749,11 @@ class ReplicaManager(val config: KafkaConfig,
* @param responseCallback callback for sending the response
* @param delayedProduceLock lock for the delayed actions
* @param recordValidationStatsCallback callback for updating stats on record conversions
- * @param requestLocal container for the stateful instances scoped to this request
- * @param transactionalId transactional ID if the request is from a producer and the producer is transactional
+ * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
+ * thread calling this method
* @param actionQueue the action queue to use. ReplicaManager#defaultActionQueue is used by default.
+ * @param verificationGuards the mapping from topic partition to verification guards if transaction verification is used
+ * @param preAppendErrors the mapping from topic partition to LogAppendResult for errors that occurred before appending
Review Comment:
This is updated.
--
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-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15087:
URL: https://github.com/apache/kafka/pull/15087#discussion_r1457569421
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -717,17 +717,16 @@ class KafkaApis(val requestChannel: RequestChannel,
val internalTopicsAllowed = request.header.clientId == AdminUtils.ADMIN_CLIENT_ID
// call the replica manager to append messages to the replicas
- replicaManager.appendRecords(
+ replicaManager.handleProduceAppend(
timeout = produceRequest.timeout.toLong,
requiredAcks = produceRequest.acks,
internalTopicsAllowed = internalTopicsAllowed,
origin = AppendOrigin.CLIENT,
Review Comment:
nit: I wonder if we should remove the `origin` parameter from `handleProduceAppend` as it should always come from a client by definition.
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -3041,31 +3007,57 @@ class ReplicaManagerTest {
origin = origin,
entriesPerPartition = Map(partition -> records),
responseCallback = appendCallback,
- transactionalId = transactionalId,
)
result
}
- private def appendRecordsToMultipleTopics(replicaManager: ReplicaManager,
- entriesToAppend: Map[TopicPartition, MemoryRecords],
- transactionalId: String,
- origin: AppendOrigin = AppendOrigin.CLIENT,
- requiredAcks: Short = -1): CallbackResult[Map[TopicPartition, PartitionResponse]] = {
+ private def handleProduceAppendToMultipleTopics(replicaManager: ReplicaManager,
+ entriesToAppend: Map[TopicPartition, MemoryRecords],
+ transactionalId: String,
+ origin: AppendOrigin = AppendOrigin.CLIENT,
+ requiredAcks: Short = -1): CallbackResult[Map[TopicPartition, PartitionResponse]] = {
val result = new CallbackResult[Map[TopicPartition, PartitionResponse]]()
def appendCallback(responses: Map[TopicPartition, PartitionResponse]): Unit = {
responses.foreach( response => assertTrue(responses.get(response._1).isDefined))
result.fire(responses)
}
+ replicaManager.handleProduceAppend(
Review Comment:
nit: Indentation seems off here.
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1188,18 +1073,7 @@ class ReplicaManager(val config: KafkaConfig,
requestLocal: RequestLocal,
verificationErrors: Map[TopicPartition, Errors]
): Unit = {
- // Map transaction coordinator errors to known errors for the response
- val convertedErrors = verificationErrors.map { case (tp, error) =>
- error match {
- case Errors.CONCURRENT_TRANSACTIONS |
- Errors.COORDINATOR_LOAD_IN_PROGRESS |
- Errors.COORDINATOR_NOT_AVAILABLE |
- Errors.NOT_COORDINATOR => tp -> Errors.NOT_ENOUGH_REPLICAS
- case _ => tp -> error
- }
-
- }
Review Comment:
For my understanding, we remove this here and we adds it back in handleProduceAppend and we rely on the conversion in the group coordinator. Did I get it right? In the group coordinator, we don't handle `CONCURRENT_TRANSACTIONS`, I think. I need to double check.
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2560,41 +2562,6 @@ class ReplicaManagerTest {
}
}
- @ParameterizedTest
- @EnumSource(value = classOf[Errors], names = Array("NOT_COORDINATOR", "CONCURRENT_TRANSACTIONS", "COORDINATOR_LOAD_IN_PROGRESS", "COORDINATOR_NOT_AVAILABLE"))
- def testMaybeVerificationErrorConversions(error: Errors): Unit = {
Review Comment:
Don't we need to keep this one as we still have those conversion but in a different place now?
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -762,167 +763,124 @@ class ReplicaManager(val config: KafkaConfig,
delayedProduceLock: Option[Lock] = None,
recordValidationStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit = _ => (),
requestLocal: RequestLocal = RequestLocal.NoCaching,
- transactionalId: String = null,
- actionQueue: ActionQueue = this.defaultActionQueue): Unit = {
- if (isValidRequiredAcks(requiredAcks)) {
-
- val verificationGuards: mutable.Map[TopicPartition, VerificationGuard] = mutable.Map[TopicPartition, VerificationGuard]()
- val (verifiedEntriesPerPartition, notYetVerifiedEntriesPerPartition, errorsPerPartition) =
- if (transactionalId == null || !config.transactionPartitionVerificationEnable)
- (entriesPerPartition, Map.empty[TopicPartition, MemoryRecords], Map.empty[TopicPartition, Errors])
- else {
- val verifiedEntries = mutable.Map[TopicPartition, MemoryRecords]()
- val unverifiedEntries = mutable.Map[TopicPartition, MemoryRecords]()
- val errorEntries = mutable.Map[TopicPartition, Errors]()
- partitionEntriesForVerification(verificationGuards, entriesPerPartition, verifiedEntries, unverifiedEntries, errorEntries)
- (verifiedEntries.toMap, unverifiedEntries.toMap, errorEntries.toMap)
- }
-
- if (notYetVerifiedEntriesPerPartition.isEmpty || addPartitionsToTxnManager.isEmpty) {
- appendEntries(verifiedEntriesPerPartition, internalTopicsAllowed, origin, requiredAcks, verificationGuards.toMap,
- errorsPerPartition, recordValidationStatsCallback, timeout, responseCallback, delayedProduceLock, actionQueue)(requestLocal, Map.empty)
- } else {
- // For unverified entries, send a request to verify. When verified, the append process will proceed via the callback.
- // We verify above that all partitions use the same producer ID.
- val batchInfo = notYetVerifiedEntriesPerPartition.head._2.firstBatch()
- addPartitionsToTxnManager.foreach(_.verifyTransaction(
- transactionalId = transactionalId,
- producerId = batchInfo.producerId,
- producerEpoch = batchInfo.producerEpoch,
- topicPartitions = notYetVerifiedEntriesPerPartition.keySet.toSeq,
- callback = KafkaRequestHandler.wrapAsyncCallback(
- appendEntries(
- entriesPerPartition,
- internalTopicsAllowed,
- origin,
- requiredAcks,
- verificationGuards.toMap,
- errorsPerPartition,
- recordValidationStatsCallback,
- timeout,
- responseCallback,
- delayedProduceLock,
- actionQueue
- ),
- requestLocal)
- ))
- }
- } else {
- // If required.acks is outside accepted range, something is wrong with the client
- // Just return an error and don't handle the request at all
- val responseStatus = entriesPerPartition.map { case (topicPartition, _) =>
- topicPartition -> new PartitionResponse(
- Errors.INVALID_REQUIRED_ACKS,
- LogAppendInfo.UNKNOWN_LOG_APPEND_INFO.firstOffset,
- RecordBatch.NO_TIMESTAMP,
- LogAppendInfo.UNKNOWN_LOG_APPEND_INFO.logStartOffset
- )
- }
- responseCallback(responseStatus)
+ actionQueue: ActionQueue = this.defaultActionQueue,
+ verificationGuards: Map[TopicPartition, VerificationGuard] = Map.empty): Unit = {
+ if (!isValidRequiredAcks(requiredAcks)) {
+ sendInvalidRequiredAcksResponse(entriesPerPartition, responseCallback)
+ return
}
- }
- /*
- * Note: This method can be used as a callback in a different request thread. Ensure that correct RequestLocal
- * is passed when executing this method. Accessing non-thread-safe data structures should be avoided if possible.
- */
- private def appendEntries(allEntries: Map[TopicPartition, MemoryRecords],
- internalTopicsAllowed: Boolean,
- origin: AppendOrigin,
- requiredAcks: Short,
- verificationGuards: Map[TopicPartition, VerificationGuard],
- errorsPerPartition: Map[TopicPartition, Errors],
- recordConversionStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit,
- timeout: Long,
- responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
- delayedProduceLock: Option[Lock],
- actionQueue: ActionQueue)
- (requestLocal: RequestLocal, unverifiedEntries: Map[TopicPartition, Errors]): Unit = {
val sTime = time.milliseconds
- val verifiedEntries =
- if (unverifiedEntries.isEmpty)
- allEntries
- else
- allEntries.filter { case (tp, _) =>
- !unverifiedEntries.contains(tp)
- }
-
val localProduceResults = appendToLocalLog(internalTopicsAllowed = internalTopicsAllowed,
- origin, verifiedEntries, requiredAcks, requestLocal, verificationGuards.toMap)
+ origin, entriesPerPartition, requiredAcks, requestLocal, verificationGuards.toMap)
debug("Produce to local log in %d ms".format(time.milliseconds - sTime))
- val errorResults = (unverifiedEntries ++ errorsPerPartition).map {
- case (topicPartition, error) =>
- // translate transaction coordinator errors to known producer response errors
- val customException =
- error match {
- case Errors.INVALID_TXN_STATE => Some(error.exception("Partition was not added to the transaction"))
- case Errors.CONCURRENT_TRANSACTIONS |
- Errors.COORDINATOR_LOAD_IN_PROGRESS |
- Errors.COORDINATOR_NOT_AVAILABLE |
- Errors.NOT_COORDINATOR => Some(new NotEnoughReplicasException(
- s"Unable to verify the partition has been added to the transaction. Underlying error: ${error.toString}"))
- case _ => None
- }
- topicPartition -> LogAppendResult(
- LogAppendInfo.UNKNOWN_LOG_APPEND_INFO,
- Some(customException.getOrElse(error.exception)),
- hasCustomErrorMessage = customException.isDefined
- )
- }
-
- val allResults = localProduceResults ++ errorResults
- val produceStatus = buildProducePartitionStatus(allResults)
+ val produceStatus = buildProducePartitionStatus(localProduceResults)
- addCompletePurgatoryAction(actionQueue, allResults)
- recordConversionStatsCallback(localProduceResults.map { case (k, v) =>
+ addCompletePurgatoryAction(actionQueue, localProduceResults)
+ recordValidationStatsCallback(localProduceResults.map { case (k, v) =>
k -> v.info.recordValidationStats
})
maybeAddDelayedProduce(
requiredAcks,
delayedProduceLock,
timeout,
- allEntries,
- allResults,
+ entriesPerPartition,
+ localProduceResults,
produceStatus,
responseCallback
)
}
- private def partitionEntriesForVerification(verificationGuards: mutable.Map[TopicPartition, VerificationGuard],
- entriesPerPartition: Map[TopicPartition, MemoryRecords],
- verifiedEntries: mutable.Map[TopicPartition, MemoryRecords],
- unverifiedEntries: mutable.Map[TopicPartition, MemoryRecords],
- errorEntries: mutable.Map[TopicPartition, Errors]): Unit = {
- val transactionalProducerIds = mutable.HashSet[Long]()
+ /**
+ * Handles the produce request by starting any transactional verification before appending.
+ *
+ * @param timeout maximum time we will wait to append before returning
+ * @param requiredAcks number of replicas who must acknowledge the append before sending the response
+ * @param internalTopicsAllowed boolean indicating whether internal topics can be appended to
+ * @param origin source of the append request (ie, client, replication, coordinator)
+ * @param transactionalId the transactional ID for the produce request or null if there is none.
+ * @param entriesPerPartition the records per partition to be appended
+ * @param responseCallback callback for sending the response
+ * @param recordValidationStatsCallback callback for updating stats on record conversions
+ * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
+ * thread calling this method
+ * @param actionQueue the action queue to use. ReplicaManager#defaultActionQueue is used by default.
+ */
+ def handleProduceAppend(timeout: Long,
+ requiredAcks: Short,
+ internalTopicsAllowed: Boolean,
+ origin: AppendOrigin,
+ transactionalId: String,
+ entriesPerPartition: Map[TopicPartition, MemoryRecords],
+ responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
+ recordValidationStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit = _ => (),
+ requestLocal: RequestLocal = RequestLocal.NoCaching,
+ actionQueue: ActionQueue = this.defaultActionQueue): Unit = {
+
+ val transactionalProducerInfo = mutable.HashSet[(Long, Short)]()
+ val topicPartitionBatchInfo = mutable.Map[TopicPartition, Int]()
entriesPerPartition.foreach { case (topicPartition, records) =>
- try {
- // Produce requests (only requests that require verification) should only have one batch per partition in "batches" but check all just to be safe.
- val transactionalBatches = records.batches.asScala.filter(batch => batch.hasProducerId && batch.isTransactional)
- transactionalBatches.foreach(batch => transactionalProducerIds.add(batch.producerId))
-
- if (transactionalBatches.nonEmpty) {
- // We return VerificationGuard if the partition needs to be verified. If no state is present, no need to verify.
- val firstBatch = records.firstBatch
- val verificationGuard = getPartitionOrException(topicPartition).maybeStartTransactionVerification(firstBatch.producerId, firstBatch.baseSequence, firstBatch.producerEpoch)
- if (verificationGuard != VerificationGuard.SENTINEL) {
- verificationGuards.put(topicPartition, verificationGuard)
- unverifiedEntries.put(topicPartition, records)
- } else
- verifiedEntries.put(topicPartition, records)
- } else {
- // If there is no producer ID or transactional records in the batches, no need to verify.
- verifiedEntries.put(topicPartition, records)
- }
- } catch {
- case e: Exception => errorEntries.put(topicPartition, Errors.forException(e))
- }
+ // Produce requests (only requests that require verification) should only have one batch per partition in "batches" but check all just to be safe.
+ val transactionalBatches = records.batches.asScala.filter(batch => batch.hasProducerId && batch.isTransactional)
+ transactionalBatches.foreach(batch => transactionalProducerInfo.add(batch.producerId, batch.producerEpoch))
+ if (!transactionalBatches.isEmpty) topicPartitionBatchInfo.put(topicPartition, records.firstBatch.baseSequence)
}
- // We should have exactly one producer ID for transactional records
- if (transactionalProducerIds.size > 1) {
+ if (transactionalProducerInfo.size > 1) {
throw new InvalidPidMappingException("Transactional records contained more than one producer ID")
}
+
+ def postVerificationCallback(preAppendErrors: Map[TopicPartition, Errors],
+ newRequestLocal: RequestLocal,
+ verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
+ val errorResults = preAppendErrors.map {
+ case (topicPartition, error) =>
+ // translate transaction coordinator errors to known producer response errors
+ val customException =
+ error match {
+ case Errors.INVALID_TXN_STATE => Some(error.exception("Partition was not added to the transaction"))
+ case Errors.CONCURRENT_TRANSACTIONS |
+ Errors.COORDINATOR_LOAD_IN_PROGRESS |
+ Errors.COORDINATOR_NOT_AVAILABLE |
+ Errors.NOT_COORDINATOR => Some(new NotEnoughReplicasException(
+ s"Unable to verify the partition has been added to the transaction. Underlying error: ${error.toString}"))
+ case _ => None
+ }
+ topicPartition -> LogAppendResult(
+ LogAppendInfo.UNKNOWN_LOG_APPEND_INFO,
+ Some(customException.getOrElse(error.exception)),
+ hasCustomErrorMessage = customException.isDefined
+ )
+ }
+ val entriesWithoutErrorsPerPartition = entriesPerPartition.filter { case (key, _) => !errorResults.contains(key) }
+
+ val preAppendPartitionResponses = buildProducePartitionStatus(errorResults).map { case (k, status) => k -> status.responseStatus }
+
+ def newResponseCallback(responses: Map[TopicPartition, PartitionResponse]): Unit = {
+ responseCallback(preAppendPartitionResponses ++ responses)
+ }
+
+ appendRecords(
+ timeout = timeout,
+ requiredAcks = requiredAcks,
+ internalTopicsAllowed = internalTopicsAllowed,
+ origin = origin,
+ entriesPerPartition = entriesWithoutErrorsPerPartition,
+ responseCallback = newResponseCallback,
+ recordValidationStatsCallback = recordValidationStatsCallback,
+ requestLocal = newRequestLocal,
+ actionQueue = actionQueue,
+ verificationGuards = verificationGuards
+ )
+ }
+
+ if (transactionalProducerInfo.size < 1) {
+ postVerificationCallback(Map.empty[TopicPartition, Errors], requestLocal, Map.empty[TopicPartition, VerificationGuard])
+ return
+ }
+ maybeStartTransactionVerificationForPartitions(topicPartitionBatchInfo, transactionalId,
Review Comment:
nit: Should we put an empty line before this one?
--
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-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #15087:
URL: https://github.com/apache/kafka/pull/15087#discussion_r2300529551
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -762,167 +763,123 @@ class ReplicaManager(val config: KafkaConfig,
delayedProduceLock: Option[Lock] = None,
recordValidationStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit = _ => (),
requestLocal: RequestLocal = RequestLocal.NoCaching,
- transactionalId: String = null,
- actionQueue: ActionQueue = this.defaultActionQueue): Unit = {
- if (isValidRequiredAcks(requiredAcks)) {
-
- val verificationGuards: mutable.Map[TopicPartition, VerificationGuard] = mutable.Map[TopicPartition, VerificationGuard]()
- val (verifiedEntriesPerPartition, notYetVerifiedEntriesPerPartition, errorsPerPartition) =
- if (transactionalId == null || !config.transactionPartitionVerificationEnable)
- (entriesPerPartition, Map.empty[TopicPartition, MemoryRecords], Map.empty[TopicPartition, Errors])
- else {
- val verifiedEntries = mutable.Map[TopicPartition, MemoryRecords]()
- val unverifiedEntries = mutable.Map[TopicPartition, MemoryRecords]()
- val errorEntries = mutable.Map[TopicPartition, Errors]()
- partitionEntriesForVerification(verificationGuards, entriesPerPartition, verifiedEntries, unverifiedEntries, errorEntries)
- (verifiedEntries.toMap, unverifiedEntries.toMap, errorEntries.toMap)
- }
-
- if (notYetVerifiedEntriesPerPartition.isEmpty || addPartitionsToTxnManager.isEmpty) {
- appendEntries(verifiedEntriesPerPartition, internalTopicsAllowed, origin, requiredAcks, verificationGuards.toMap,
- errorsPerPartition, recordValidationStatsCallback, timeout, responseCallback, delayedProduceLock, actionQueue)(requestLocal, Map.empty)
- } else {
- // For unverified entries, send a request to verify. When verified, the append process will proceed via the callback.
- // We verify above that all partitions use the same producer ID.
- val batchInfo = notYetVerifiedEntriesPerPartition.head._2.firstBatch()
- addPartitionsToTxnManager.foreach(_.verifyTransaction(
- transactionalId = transactionalId,
- producerId = batchInfo.producerId,
- producerEpoch = batchInfo.producerEpoch,
- topicPartitions = notYetVerifiedEntriesPerPartition.keySet.toSeq,
- callback = KafkaRequestHandler.wrapAsyncCallback(
- appendEntries(
- entriesPerPartition,
- internalTopicsAllowed,
- origin,
- requiredAcks,
- verificationGuards.toMap,
- errorsPerPartition,
- recordValidationStatsCallback,
- timeout,
- responseCallback,
- delayedProduceLock,
- actionQueue
- ),
- requestLocal)
- ))
- }
- } else {
- // If required.acks is outside accepted range, something is wrong with the client
- // Just return an error and don't handle the request at all
- val responseStatus = entriesPerPartition.map { case (topicPartition, _) =>
- topicPartition -> new PartitionResponse(
- Errors.INVALID_REQUIRED_ACKS,
- LogAppendInfo.UNKNOWN_LOG_APPEND_INFO.firstOffset,
- RecordBatch.NO_TIMESTAMP,
- LogAppendInfo.UNKNOWN_LOG_APPEND_INFO.logStartOffset
- )
- }
- responseCallback(responseStatus)
+ actionQueue: ActionQueue = this.defaultActionQueue,
+ verificationGuards: Map[TopicPartition, VerificationGuard] = Map.empty): Unit = {
+ if (!isValidRequiredAcks(requiredAcks)) {
+ sendInvalidRequiredAcksResponse(entriesPerPartition, responseCallback)
+ return
}
- }
- /*
- * Note: This method can be used as a callback in a different request thread. Ensure that correct RequestLocal
- * is passed when executing this method. Accessing non-thread-safe data structures should be avoided if possible.
- */
- private def appendEntries(allEntries: Map[TopicPartition, MemoryRecords],
- internalTopicsAllowed: Boolean,
- origin: AppendOrigin,
- requiredAcks: Short,
- verificationGuards: Map[TopicPartition, VerificationGuard],
- errorsPerPartition: Map[TopicPartition, Errors],
- recordConversionStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit,
- timeout: Long,
- responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
- delayedProduceLock: Option[Lock],
- actionQueue: ActionQueue)
- (requestLocal: RequestLocal, unverifiedEntries: Map[TopicPartition, Errors]): Unit = {
val sTime = time.milliseconds
- val verifiedEntries =
- if (unverifiedEntries.isEmpty)
- allEntries
- else
- allEntries.filter { case (tp, _) =>
- !unverifiedEntries.contains(tp)
- }
-
val localProduceResults = appendToLocalLog(internalTopicsAllowed = internalTopicsAllowed,
- origin, verifiedEntries, requiredAcks, requestLocal, verificationGuards.toMap)
+ origin, entriesPerPartition, requiredAcks, requestLocal, verificationGuards.toMap)
debug("Produce to local log in %d ms".format(time.milliseconds - sTime))
- val errorResults = (unverifiedEntries ++ errorsPerPartition).map {
- case (topicPartition, error) =>
- // translate transaction coordinator errors to known producer response errors
- val customException =
- error match {
- case Errors.INVALID_TXN_STATE => Some(error.exception("Partition was not added to the transaction"))
- case Errors.CONCURRENT_TRANSACTIONS |
- Errors.COORDINATOR_LOAD_IN_PROGRESS |
- Errors.COORDINATOR_NOT_AVAILABLE |
- Errors.NOT_COORDINATOR => Some(new NotEnoughReplicasException(
- s"Unable to verify the partition has been added to the transaction. Underlying error: ${error.toString}"))
- case _ => None
- }
- topicPartition -> LogAppendResult(
- LogAppendInfo.UNKNOWN_LOG_APPEND_INFO,
- Some(customException.getOrElse(error.exception)),
- hasCustomErrorMessage = customException.isDefined
- )
- }
-
- val allResults = localProduceResults ++ errorResults
- val produceStatus = buildProducePartitionStatus(allResults)
+ val produceStatus = buildProducePartitionStatus(localProduceResults)
- addCompletePurgatoryAction(actionQueue, allResults)
- recordConversionStatsCallback(localProduceResults.map { case (k, v) =>
+ addCompletePurgatoryAction(actionQueue, localProduceResults)
+ recordValidationStatsCallback(localProduceResults.map { case (k, v) =>
k -> v.info.recordValidationStats
})
maybeAddDelayedProduce(
requiredAcks,
delayedProduceLock,
timeout,
- allEntries,
- allResults,
+ entriesPerPartition,
+ localProduceResults,
produceStatus,
responseCallback
)
}
- private def partitionEntriesForVerification(verificationGuards: mutable.Map[TopicPartition, VerificationGuard],
- entriesPerPartition: Map[TopicPartition, MemoryRecords],
- verifiedEntries: mutable.Map[TopicPartition, MemoryRecords],
- unverifiedEntries: mutable.Map[TopicPartition, MemoryRecords],
- errorEntries: mutable.Map[TopicPartition, Errors]): Unit = {
- val transactionalProducerIds = mutable.HashSet[Long]()
+ /**
+ * Handles the produce request by starting any transactional verification before appending.
+ *
+ * @param timeout maximum time we will wait to append before returning
+ * @param requiredAcks number of replicas who must acknowledge the append before sending the response
+ * @param internalTopicsAllowed boolean indicating whether internal topics can be appended to
+ * @param transactionalId the transactional ID for the produce request or null if there is none.
+ * @param entriesPerPartition the records per partition to be appended
+ * @param responseCallback callback for sending the response
+ * @param recordValidationStatsCallback callback for updating stats on record conversions
+ * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
+ * thread calling this method
+ * @param actionQueue the action queue to use. ReplicaManager#defaultActionQueue is used by default.
+ */
+ def handleProduceAppend(timeout: Long,
+ requiredAcks: Short,
+ internalTopicsAllowed: Boolean,
+ transactionalId: String,
+ entriesPerPartition: Map[TopicPartition, MemoryRecords],
+ responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
+ recordValidationStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit = _ => (),
+ requestLocal: RequestLocal = RequestLocal.NoCaching,
+ actionQueue: ActionQueue = this.defaultActionQueue): Unit = {
+
+ val transactionalProducerInfo = mutable.HashSet[(Long, Short)]()
+ val topicPartitionBatchInfo = mutable.Map[TopicPartition, Int]()
entriesPerPartition.foreach { case (topicPartition, records) =>
- try {
- // Produce requests (only requests that require verification) should only have one batch per partition in "batches" but check all just to be safe.
- val transactionalBatches = records.batches.asScala.filter(batch => batch.hasProducerId && batch.isTransactional)
- transactionalBatches.foreach(batch => transactionalProducerIds.add(batch.producerId))
-
- if (transactionalBatches.nonEmpty) {
- // We return VerificationGuard if the partition needs to be verified. If no state is present, no need to verify.
- val firstBatch = records.firstBatch
- val verificationGuard = getPartitionOrException(topicPartition).maybeStartTransactionVerification(firstBatch.producerId, firstBatch.baseSequence, firstBatch.producerEpoch)
- if (verificationGuard != VerificationGuard.SENTINEL) {
- verificationGuards.put(topicPartition, verificationGuard)
- unverifiedEntries.put(topicPartition, records)
- } else
- verifiedEntries.put(topicPartition, records)
- } else {
- // If there is no producer ID or transactional records in the batches, no need to verify.
- verifiedEntries.put(topicPartition, records)
- }
- } catch {
- case e: Exception => errorEntries.put(topicPartition, Errors.forException(e))
- }
+ // Produce requests (only requests that require verification) should only have one batch per partition in "batches" but check all just to be safe.
+ val transactionalBatches = records.batches.asScala.filter(batch => batch.hasProducerId && batch.isTransactional)
+ transactionalBatches.foreach(batch => transactionalProducerInfo.add(batch.producerId, batch.producerEpoch))
+ if (!transactionalBatches.isEmpty) topicPartitionBatchInfo.put(topicPartition, records.firstBatch.baseSequence)
}
- // We should have exactly one producer ID for transactional records
- if (transactionalProducerIds.size > 1) {
+ if (transactionalProducerInfo.size > 1) {
throw new InvalidPidMappingException("Transactional records contained more than one producer ID")
}
+
+ def postVerificationCallback(preAppendErrors: Map[TopicPartition, Errors],
+ newRequestLocal: RequestLocal,
+ verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
+ val errorResults = preAppendErrors.map {
+ case (topicPartition, error) =>
+ // translate transaction coordinator errors to known producer response errors
+ val customException =
+ error match {
+ case Errors.INVALID_TXN_STATE => Some(error.exception("Partition was not added to the transaction"))
+ case Errors.CONCURRENT_TRANSACTIONS |
+ Errors.COORDINATOR_LOAD_IN_PROGRESS |
+ Errors.COORDINATOR_NOT_AVAILABLE |
+ Errors.NOT_COORDINATOR => Some(new NotEnoughReplicasException(
+ s"Unable to verify the partition has been added to the transaction. Underlying error: ${error.toString}"))
+ case _ => None
+ }
+ topicPartition -> LogAppendResult(
+ LogAppendInfo.UNKNOWN_LOG_APPEND_INFO,
+ Some(customException.getOrElse(error.exception)),
+ hasCustomErrorMessage = customException.isDefined
+ )
+ }
+ val entriesWithoutErrorsPerPartition = entriesPerPartition.filter { case (key, _) => !errorResults.contains(key) }
Review Comment:
I open the PR to addressed this improvement https://github.com/apache/kafka/pull/20410
--
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-15987: Refactor ReplicaManager code for transaction verification [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15087:
URL: https://github.com/apache/kafka/pull/15087#discussion_r2298430334
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -762,167 +763,123 @@ class ReplicaManager(val config: KafkaConfig,
delayedProduceLock: Option[Lock] = None,
recordValidationStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit = _ => (),
requestLocal: RequestLocal = RequestLocal.NoCaching,
- transactionalId: String = null,
- actionQueue: ActionQueue = this.defaultActionQueue): Unit = {
- if (isValidRequiredAcks(requiredAcks)) {
-
- val verificationGuards: mutable.Map[TopicPartition, VerificationGuard] = mutable.Map[TopicPartition, VerificationGuard]()
- val (verifiedEntriesPerPartition, notYetVerifiedEntriesPerPartition, errorsPerPartition) =
- if (transactionalId == null || !config.transactionPartitionVerificationEnable)
- (entriesPerPartition, Map.empty[TopicPartition, MemoryRecords], Map.empty[TopicPartition, Errors])
- else {
- val verifiedEntries = mutable.Map[TopicPartition, MemoryRecords]()
- val unverifiedEntries = mutable.Map[TopicPartition, MemoryRecords]()
- val errorEntries = mutable.Map[TopicPartition, Errors]()
- partitionEntriesForVerification(verificationGuards, entriesPerPartition, verifiedEntries, unverifiedEntries, errorEntries)
- (verifiedEntries.toMap, unverifiedEntries.toMap, errorEntries.toMap)
- }
-
- if (notYetVerifiedEntriesPerPartition.isEmpty || addPartitionsToTxnManager.isEmpty) {
- appendEntries(verifiedEntriesPerPartition, internalTopicsAllowed, origin, requiredAcks, verificationGuards.toMap,
- errorsPerPartition, recordValidationStatsCallback, timeout, responseCallback, delayedProduceLock, actionQueue)(requestLocal, Map.empty)
- } else {
- // For unverified entries, send a request to verify. When verified, the append process will proceed via the callback.
- // We verify above that all partitions use the same producer ID.
- val batchInfo = notYetVerifiedEntriesPerPartition.head._2.firstBatch()
- addPartitionsToTxnManager.foreach(_.verifyTransaction(
- transactionalId = transactionalId,
- producerId = batchInfo.producerId,
- producerEpoch = batchInfo.producerEpoch,
- topicPartitions = notYetVerifiedEntriesPerPartition.keySet.toSeq,
- callback = KafkaRequestHandler.wrapAsyncCallback(
- appendEntries(
- entriesPerPartition,
- internalTopicsAllowed,
- origin,
- requiredAcks,
- verificationGuards.toMap,
- errorsPerPartition,
- recordValidationStatsCallback,
- timeout,
- responseCallback,
- delayedProduceLock,
- actionQueue
- ),
- requestLocal)
- ))
- }
- } else {
- // If required.acks is outside accepted range, something is wrong with the client
- // Just return an error and don't handle the request at all
- val responseStatus = entriesPerPartition.map { case (topicPartition, _) =>
- topicPartition -> new PartitionResponse(
- Errors.INVALID_REQUIRED_ACKS,
- LogAppendInfo.UNKNOWN_LOG_APPEND_INFO.firstOffset,
- RecordBatch.NO_TIMESTAMP,
- LogAppendInfo.UNKNOWN_LOG_APPEND_INFO.logStartOffset
- )
- }
- responseCallback(responseStatus)
+ actionQueue: ActionQueue = this.defaultActionQueue,
+ verificationGuards: Map[TopicPartition, VerificationGuard] = Map.empty): Unit = {
+ if (!isValidRequiredAcks(requiredAcks)) {
+ sendInvalidRequiredAcksResponse(entriesPerPartition, responseCallback)
+ return
}
- }
- /*
- * Note: This method can be used as a callback in a different request thread. Ensure that correct RequestLocal
- * is passed when executing this method. Accessing non-thread-safe data structures should be avoided if possible.
- */
- private def appendEntries(allEntries: Map[TopicPartition, MemoryRecords],
- internalTopicsAllowed: Boolean,
- origin: AppendOrigin,
- requiredAcks: Short,
- verificationGuards: Map[TopicPartition, VerificationGuard],
- errorsPerPartition: Map[TopicPartition, Errors],
- recordConversionStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit,
- timeout: Long,
- responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
- delayedProduceLock: Option[Lock],
- actionQueue: ActionQueue)
- (requestLocal: RequestLocal, unverifiedEntries: Map[TopicPartition, Errors]): Unit = {
val sTime = time.milliseconds
- val verifiedEntries =
- if (unverifiedEntries.isEmpty)
- allEntries
- else
- allEntries.filter { case (tp, _) =>
- !unverifiedEntries.contains(tp)
- }
-
val localProduceResults = appendToLocalLog(internalTopicsAllowed = internalTopicsAllowed,
- origin, verifiedEntries, requiredAcks, requestLocal, verificationGuards.toMap)
+ origin, entriesPerPartition, requiredAcks, requestLocal, verificationGuards.toMap)
debug("Produce to local log in %d ms".format(time.milliseconds - sTime))
- val errorResults = (unverifiedEntries ++ errorsPerPartition).map {
- case (topicPartition, error) =>
- // translate transaction coordinator errors to known producer response errors
- val customException =
- error match {
- case Errors.INVALID_TXN_STATE => Some(error.exception("Partition was not added to the transaction"))
- case Errors.CONCURRENT_TRANSACTIONS |
- Errors.COORDINATOR_LOAD_IN_PROGRESS |
- Errors.COORDINATOR_NOT_AVAILABLE |
- Errors.NOT_COORDINATOR => Some(new NotEnoughReplicasException(
- s"Unable to verify the partition has been added to the transaction. Underlying error: ${error.toString}"))
- case _ => None
- }
- topicPartition -> LogAppendResult(
- LogAppendInfo.UNKNOWN_LOG_APPEND_INFO,
- Some(customException.getOrElse(error.exception)),
- hasCustomErrorMessage = customException.isDefined
- )
- }
-
- val allResults = localProduceResults ++ errorResults
- val produceStatus = buildProducePartitionStatus(allResults)
+ val produceStatus = buildProducePartitionStatus(localProduceResults)
- addCompletePurgatoryAction(actionQueue, allResults)
- recordConversionStatsCallback(localProduceResults.map { case (k, v) =>
+ addCompletePurgatoryAction(actionQueue, localProduceResults)
+ recordValidationStatsCallback(localProduceResults.map { case (k, v) =>
k -> v.info.recordValidationStats
})
maybeAddDelayedProduce(
requiredAcks,
delayedProduceLock,
timeout,
- allEntries,
- allResults,
+ entriesPerPartition,
+ localProduceResults,
produceStatus,
responseCallback
)
}
- private def partitionEntriesForVerification(verificationGuards: mutable.Map[TopicPartition, VerificationGuard],
- entriesPerPartition: Map[TopicPartition, MemoryRecords],
- verifiedEntries: mutable.Map[TopicPartition, MemoryRecords],
- unverifiedEntries: mutable.Map[TopicPartition, MemoryRecords],
- errorEntries: mutable.Map[TopicPartition, Errors]): Unit = {
- val transactionalProducerIds = mutable.HashSet[Long]()
+ /**
+ * Handles the produce request by starting any transactional verification before appending.
+ *
+ * @param timeout maximum time we will wait to append before returning
+ * @param requiredAcks number of replicas who must acknowledge the append before sending the response
+ * @param internalTopicsAllowed boolean indicating whether internal topics can be appended to
+ * @param transactionalId the transactional ID for the produce request or null if there is none.
+ * @param entriesPerPartition the records per partition to be appended
+ * @param responseCallback callback for sending the response
+ * @param recordValidationStatsCallback callback for updating stats on record conversions
+ * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the
+ * thread calling this method
+ * @param actionQueue the action queue to use. ReplicaManager#defaultActionQueue is used by default.
+ */
+ def handleProduceAppend(timeout: Long,
+ requiredAcks: Short,
+ internalTopicsAllowed: Boolean,
+ transactionalId: String,
+ entriesPerPartition: Map[TopicPartition, MemoryRecords],
+ responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
+ recordValidationStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit = _ => (),
+ requestLocal: RequestLocal = RequestLocal.NoCaching,
+ actionQueue: ActionQueue = this.defaultActionQueue): Unit = {
+
+ val transactionalProducerInfo = mutable.HashSet[(Long, Short)]()
+ val topicPartitionBatchInfo = mutable.Map[TopicPartition, Int]()
entriesPerPartition.foreach { case (topicPartition, records) =>
- try {
- // Produce requests (only requests that require verification) should only have one batch per partition in "batches" but check all just to be safe.
- val transactionalBatches = records.batches.asScala.filter(batch => batch.hasProducerId && batch.isTransactional)
- transactionalBatches.foreach(batch => transactionalProducerIds.add(batch.producerId))
-
- if (transactionalBatches.nonEmpty) {
- // We return VerificationGuard if the partition needs to be verified. If no state is present, no need to verify.
- val firstBatch = records.firstBatch
- val verificationGuard = getPartitionOrException(topicPartition).maybeStartTransactionVerification(firstBatch.producerId, firstBatch.baseSequence, firstBatch.producerEpoch)
- if (verificationGuard != VerificationGuard.SENTINEL) {
- verificationGuards.put(topicPartition, verificationGuard)
- unverifiedEntries.put(topicPartition, records)
- } else
- verifiedEntries.put(topicPartition, records)
- } else {
- // If there is no producer ID or transactional records in the batches, no need to verify.
- verifiedEntries.put(topicPartition, records)
- }
- } catch {
- case e: Exception => errorEntries.put(topicPartition, Errors.forException(e))
- }
+ // Produce requests (only requests that require verification) should only have one batch per partition in "batches" but check all just to be safe.
+ val transactionalBatches = records.batches.asScala.filter(batch => batch.hasProducerId && batch.isTransactional)
+ transactionalBatches.foreach(batch => transactionalProducerInfo.add(batch.producerId, batch.producerEpoch))
+ if (!transactionalBatches.isEmpty) topicPartitionBatchInfo.put(topicPartition, records.firstBatch.baseSequence)
}
- // We should have exactly one producer ID for transactional records
- if (transactionalProducerIds.size > 1) {
+ if (transactionalProducerInfo.size > 1) {
throw new InvalidPidMappingException("Transactional records contained more than one producer ID")
}
+
+ def postVerificationCallback(preAppendErrors: Map[TopicPartition, Errors],
+ newRequestLocal: RequestLocal,
+ verificationGuards: Map[TopicPartition, VerificationGuard]): Unit = {
+ val errorResults = preAppendErrors.map {
+ case (topicPartition, error) =>
+ // translate transaction coordinator errors to known producer response errors
+ val customException =
+ error match {
+ case Errors.INVALID_TXN_STATE => Some(error.exception("Partition was not added to the transaction"))
+ case Errors.CONCURRENT_TRANSACTIONS |
+ Errors.COORDINATOR_LOAD_IN_PROGRESS |
+ Errors.COORDINATOR_NOT_AVAILABLE |
+ Errors.NOT_COORDINATOR => Some(new NotEnoughReplicasException(
+ s"Unable to verify the partition has been added to the transaction. Underlying error: ${error.toString}"))
+ case _ => None
+ }
+ topicPartition -> LogAppendResult(
+ LogAppendInfo.UNKNOWN_LOG_APPEND_INFO,
+ Some(customException.getOrElse(error.exception)),
+ hasCustomErrorMessage = customException.isDefined
+ )
+ }
+ val entriesWithoutErrorsPerPartition = entriesPerPartition.filter { case (key, _) => !errorResults.contains(key) }
Review Comment:
Yeah, we could probably only do this filtering if the errorResults is nonEmpty.
--
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