jolshan commented on code in PR #14774:
URL: https://github.com/apache/kafka/pull/14774#discussion_r1427224966
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -887,7 +886,63 @@ class ReplicaManager(val config: KafkaConfig,
}
val allResults = localProduceResults ++ errorResults
- val produceStatus = allResults.map { case (topicPartition, result) =>
+ val produceStatus = buildProducePartitionStatus(allResults)
+
+ addCompletePurgatoryAction(actionQueue, allResults)
+ recordConversionStatsCallback(localProduceResults.map { case (k, v) =>
+ k -> v.info.recordValidationStats
+ })
+
+ maybeAddDelayedProduce(
+ requiredAcks,
+ delayedProduceLock,
+ timeout,
+ allEntries,
+ allResults,
+ 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]()
+ 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))
+ }
+ }
+ // We should have exactly one producer ID for transactional records
+ if (transactionalProducerIds.size > 1) {
+ throw new InvalidPidMappingException("Transactional records contained
more than one producer ID")
+ }
+ }
+
+ private def buildProducePartitionStatus(
Review Comment:
All of these will be covered in the refactor. I didn't touch the produce
flow to minimize the diff and cause minimal confusion when reviewing.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
