hachikuji commented on code in PR #14774:
URL: https://github.com/apache/kafka/pull/14774#discussion_r1418127368
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -941,39 +851,209 @@ class ReplicaManager(val config: KafkaConfig,
}
}
- 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 entriesPerPartition the records per partition to be
appended
+ * @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 -- 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
+ */
+ 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
+ )
+ }
+
+ appendRecords(
+ timeout = timeout,
+ requiredAcks = requiredAcks,
+ internalTopicsAllowed = internalTopicsAllowed,
+ origin = origin,
+ entriesPerPartition = entriesPerPartition,
+ responseCallback = responseCallback,
+ recordValidationStatsCallback = recordValidationStatsCallback,
+ requestLocal = newRequestLocal,
+ verificationGuards = verificationGuards,
+ preAppendErrors = errorResults
+ )
+ }
+
+ if (transactionalProducerInfo.size < 1) {
+ postVerificationCallback(Map.empty[TopicPartition, Errors],
requestLocal, Map.empty[TopicPartition, VerificationGuard])
+ return
+ }
+ maybeStartTransactionVerificationForPartitions(topicPartitionBatchInfo,
transactionalId,
+ transactionalProducerInfo.head._1, transactionalProducerInfo.head._2,
requestLocal, postVerificationCallback)
+ }
+
+ private def sendInvalidRequiredAcksResponse(entries: Map[TopicPartition,
MemoryRecords],
+ responseCallback:
Map[TopicPartition, PartitionResponse] => Unit): Unit = {
+ // 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 = entries.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)
+ }
+
+ def maybeStartTransactionVerificationForPartition(
+ topicPartition: TopicPartition,
+ transactionalId: String,
+ producerId: Long,
+ producerEpoch: Short,
+ baseSequence: Int,
+ requestLocal: RequestLocal,
+ callback: (Errors, RequestLocal, VerificationGuard) => Unit
+ ): Unit = {
+ def generalizedCallback(preAppendErrors: Map[TopicPartition, Errors],
+ newRequestLocal: RequestLocal,
+ verificationGuards: Map[TopicPartition,
VerificationGuard]): Unit = {
+ callback(preAppendErrors.getOrElse(topicPartition, Errors.NONE),
newRequestLocal, verificationGuards.getOrElse(topicPartition,
VerificationGuard.SENTINEL))
+ }
+
+ maybeStartTransactionVerificationForPartitions(
+ Map(topicPartition -> baseSequence),
+ transactionalId,
+ producerId,
+ producerEpoch,
+ requestLocal,
+ generalizedCallback
+ )
+ }
+
+ def maybeStartTransactionVerificationForPartitions(
+ topicPartitionBatchInfo: Map[TopicPartition, Int],
+ transactionalId: String,
+ producerId: Long,
+ producerEpoch: Short,
+ requestLocal: RequestLocal,
+ callback: (Map[TopicPartition, Errors], RequestLocal, Map[TopicPartition,
VerificationGuard]) => Unit
Review Comment:
Wonder if we could simplify this. Would `Map[TopicPartition, Either[Errors,
VerificationGuard]]` work to replace the two maps?
--
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
