hachikuji commented on code in PR #13607: URL: https://github.com/apache/kafka/pull/13607#discussion_r1183173661
########## core/src/main/scala/kafka/server/ReplicaManager.scala: ########## @@ -744,6 +751,8 @@ class ReplicaManager(val config: KafkaConfig, } // map not yet verified partitions to a request object + // Since verification occurs on produce requests only, and each produce request has one batch per partition, we know the producer ID is transactional Review Comment: But do we know that the producerId is consistent among all transactional batches? Seems like we are assuming it below, but where is it verified? ########## core/src/main/scala/kafka/server/ReplicaManager.scala: ########## @@ -642,7 +642,14 @@ class ReplicaManager(val config: KafkaConfig, (entriesPerPartition, Map.empty) else entriesPerPartition.partition { case (topicPartition, records) => - getPartitionOrException(topicPartition).hasOngoingTransaction(records.firstBatch().producerId()) + // 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) + if (!transactionalBatches.isEmpty) { Review Comment: nit: `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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org