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