jolshan commented on code in PR #13231:
URL: https://github.com/apache/kafka/pull/13231#discussion_r1124831869
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala:
##########
@@ -317,6 +322,34 @@ class TransactionCoordinator(txnConfig: TransactionConfig,
}
}
}
+
+ def handleVerifyPartitionsInTransaction(transactionalId: String,
+ producerId: Long,
+ producerEpoch: Short,
+ partitions:
collection.Set[TopicPartition],
+ responseCallback:
VerifyPartitionsCallback): Unit = {
+ if (transactionalId == null || transactionalId.isEmpty) {
+ debug(s"Returning ${Errors.INVALID_REQUEST} error code to client for
$transactionalId's AddPartitions request")
+
responseCallback(AddPartitionsToTxnResponse.resultForTransaction(transactionalId,
partitions.map(_ -> Errors.INVALID_REQUEST).toMap.asJava))
+ } else {
+ val result: ApiResult[(Int, TransactionMetadata)] =
getTransactionMetadata(transactionalId, producerId, producerEpoch, partitions)
+
+ result match {
+ case Left(err) =>
+ debug(s"Returning $err error code to client for $transactionalId's
AddPartitions request")
+
responseCallback(AddPartitionsToTxnResponse.resultForTransaction(transactionalId,
partitions.map(_ -> err).toMap.asJava))
+
+ case Right((_, txnMetadata)) =>
+ val txnMetadataPartitions = txnMetadata.topicPartitions
+ val addedPartitions = partitions.intersect(txnMetadataPartitions)
+ val nonAddedPartitions = partitions.diff(txnMetadataPartitions)
+ val errors = mutable.Map[TopicPartition, Errors]()
+ addedPartitions.foreach(errors.put(_, Errors.NONE))
+ nonAddedPartitions.foreach(errors.put(_, Errors.INVALID_TXN_STATE))
Review Comment:
sure
--
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
