diff --git a/core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala b/core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala index 2c69372558f34..2e6144fe84a28 100644 --- a/core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala +++ b/core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala @@ -27,7 +27,7 @@ import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse import org.apache.kafka.common.requests.TransactionResult import org.apache.kafka.common.utils.{BufferSupplier, Time} import org.apache.kafka.coordinator.group.runtime.PartitionWriter -import org.apache.kafka.storage.internals.log.VerificationGuard +import org.apache.kafka.storage.internals.log.{AppendOrigin, VerificationGuard} import java.util import java.util.concurrent.CompletableFuture @@ -215,9 +215,11 @@ class CoordinatorPartitionWriter[T]( verificationGuard: VerificationGuard = VerificationGuard.SENTINEL ): Long = { var appendResults: Map[TopicPartition, PartitionResponse] = Map.empty - replicaManager.appendForGroup( + replicaManager.appendRecords( timeout = 0L, requiredAcks = 1, + internalTopicsAllowed = true, + origin = AppendOrigin.COORDINATOR, entriesPerPartition = Map(tp -> memoryRecords), responseCallback = results => appendResults = results, requestLocal = RequestLocal.NoCaching, diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala index 3177576eed1c1..6012722332d56 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala @@ -335,9 +335,11 @@ class GroupMetadataManager(brokerId: Int, verificationGuards: Map[TopicPartition, VerificationGuard] = Map.empty ): Unit = { // call replica manager to append the group message - replicaManager.appendForGroup( + replicaManager.appendRecords( timeout = config.offsetCommitTimeoutMs.toLong, requiredAcks = config.offsetCommitRequiredAcks, + internalTopicsAllowed = true, + origin = AppendOrigin.COORDINATOR, entriesPerPartition = records, delayedProduceLock = Some(group.lock), responseCallback = callback, diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala index af46bfcc95c63..10445dcb99443 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala @@ -281,12 +281,12 @@ class TransactionStateManager(brokerId: Int, inReadLock(stateLock) { replicaManager.appendRecords( - config.requestTimeoutMs, - TransactionLog.EnforcedRequiredAcks, + timeout = config.requestTimeoutMs, + requiredAcks = TransactionLog.EnforcedRequiredAcks, internalTopicsAllowed = true, origin = AppendOrigin.COORDINATOR, entriesPerPartition = Map(transactionPartition -> tombstoneRecords), - removeFromCacheCallback, + responseCallback = removeFromCacheCallback, requestLocal = RequestLocal.NoCaching) } } diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index 93f55f03998dc..4aac943877f8f 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -717,17 +717,15 @@ class KafkaApis(val requestChannel: RequestChannel, val internalTopicsAllowed = request.header.clientId == AdminUtils.ADMIN_CLIENT_ID // call the replica manager to append messages to the replicas - replicaManager.appendRecords( + replicaManager.handleProduceAppend( timeout = produceRequest.timeout.toLong, requiredAcks = produceRequest.acks, internalTopicsAllowed = internalTopicsAllowed, - origin = AppendOrigin.CLIENT, + transactionalId = produceRequest.transactionalId, entriesPerPartition = authorizedRequestInfo, - requestLocal = requestLocal, responseCallback = sendResponseCallback, recordValidationStatsCallback = processingStatsCallback, - transactionalId = produceRequest.transactionalId() - ) + requestLocal = requestLocal) // if the request is put into the purgatory, it will have a held reference and hence cannot be garbage collected; // hence we clear its data here in order to let GC reclaim its memory since it is already appended to log diff --git a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala index 645a33ac931d3..7f301f787075d 100755 --- a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala +++ b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala @@ -67,17 +67,17 @@ object KafkaRequestHandler { if (requestChannel == null || currentRequest == null) { if (!bypassThreadCheck) throw new IllegalStateException("Attempted to reschedule to request handler thread from non-request handler thread.") - T => asyncCompletionCallback(requestLocal, T) + t => asyncCompletionCallback(requestLocal, t) } else { - T => { + t => { if (threadCurrentRequest.get() == currentRequest) { // If the callback is actually executed on the same request thread, we can directly execute // it without re-scheduling it. - asyncCompletionCallback(requestLocal, T) + asyncCompletionCallback(requestLocal, t) } else { // The requestChannel and request are captured in this lambda, so when it's executed on the callback thread // we can re-schedule the original callback on a request thread and update the metrics accordingly. - requestChannel.sendCallbackRequest(RequestChannel.CallbackRequest(newRequestLocal => asyncCompletionCallback(newRequestLocal, T), currentRequest)) + requestChannel.sendCallbackRequest(RequestChannel.CallbackRequest(newRequestLocal => asyncCompletionCallback(newRequestLocal, t), currentRequest)) } } } diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 48ec0d0fe18aa..34eee216b9f3b 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -749,9 +749,10 @@ class ReplicaManager(val config: KafkaConfig, * @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 - * @param transactionalId transactional ID if the request is from a producer and the producer is transactional + * @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 appendRecords(timeout: Long, requiredAcks: Short, @@ -762,120 +763,22 @@ class ReplicaManager(val config: KafkaConfig, delayedProduceLock: Option[Lock] = None, recordValidationStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit = _ => (), requestLocal: RequestLocal = RequestLocal.NoCaching, - transactionalId: String = null, - actionQueue: ActionQueue = this.defaultActionQueue): Unit = { - if (isValidRequiredAcks(requiredAcks)) { - - val verificationGuards: mutable.Map[TopicPartition, VerificationGuard] = mutable.Map[TopicPartition, VerificationGuard]() - val (verifiedEntriesPerPartition, notYetVerifiedEntriesPerPartition, errorsPerPartition) = - if (transactionalId == null || !config.transactionPartitionVerificationEnable) - (entriesPerPartition, Map.empty[TopicPartition, MemoryRecords], Map.empty[TopicPartition, Errors]) - else { - val verifiedEntries = mutable.Map[TopicPartition, MemoryRecords]() - val unverifiedEntries = mutable.Map[TopicPartition, MemoryRecords]() - val errorEntries = mutable.Map[TopicPartition, Errors]() - partitionEntriesForVerification(verificationGuards, entriesPerPartition, verifiedEntries, unverifiedEntries, errorEntries) - (verifiedEntries.toMap, unverifiedEntries.toMap, errorEntries.toMap) - } - - if (notYetVerifiedEntriesPerPartition.isEmpty || addPartitionsToTxnManager.isEmpty) { - appendEntries(verifiedEntriesPerPartition, internalTopicsAllowed, origin, requiredAcks, verificationGuards.toMap, - errorsPerPartition, recordValidationStatsCallback, timeout, responseCallback, delayedProduceLock, actionQueue)(requestLocal, Map.empty) - } else { - // For unverified entries, send a request to verify. When verified, the append process will proceed via the callback. - // We verify above that all partitions use the same producer ID. - val batchInfo = notYetVerifiedEntriesPerPartition.head._2.firstBatch() - addPartitionsToTxnManager.foreach(_.verifyTransaction( - transactionalId = transactionalId, - producerId = batchInfo.producerId, - producerEpoch = batchInfo.producerEpoch, - topicPartitions = notYetVerifiedEntriesPerPartition.keySet.toSeq, - callback = KafkaRequestHandler.wrapAsyncCallback( - appendEntries( - entriesPerPartition, - internalTopicsAllowed, - origin, - requiredAcks, - verificationGuards.toMap, - errorsPerPartition, - recordValidationStatsCallback, - timeout, - responseCallback, - delayedProduceLock, - actionQueue - ), - requestLocal) - )) - } - } else { - // 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 = entriesPerPartition.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) + actionQueue: ActionQueue = this.defaultActionQueue, + verificationGuards: Map[TopicPartition, VerificationGuard] = Map.empty): Unit = { + if (!isValidRequiredAcks(requiredAcks)) { + sendInvalidRequiredAcksResponse(entriesPerPartition, responseCallback) + return } - } - /* - * Note: This method can be used as a callback in a different request thread. Ensure that correct RequestLocal - * is passed when executing this method. Accessing non-thread-safe data structures should be avoided if possible. - */ - private def appendEntries(allEntries: Map[TopicPartition, MemoryRecords], - internalTopicsAllowed: Boolean, - origin: AppendOrigin, - requiredAcks: Short, - verificationGuards: Map[TopicPartition, VerificationGuard], - errorsPerPartition: Map[TopicPartition, Errors], - recordConversionStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit, - timeout: Long, - responseCallback: Map[TopicPartition, PartitionResponse] => Unit, - delayedProduceLock: Option[Lock], - actionQueue: ActionQueue) - (requestLocal: RequestLocal, unverifiedEntries: Map[TopicPartition, Errors]): Unit = { val sTime = time.milliseconds - val verifiedEntries = - if (unverifiedEntries.isEmpty) - allEntries - else - allEntries.filter { case (tp, _) => - !unverifiedEntries.contains(tp) - } - val localProduceResults = appendToLocalLog(internalTopicsAllowed = internalTopicsAllowed, - origin, verifiedEntries, requiredAcks, requestLocal, verificationGuards.toMap) + origin, entriesPerPartition, requiredAcks, requestLocal, verificationGuards.toMap) debug("Produce to local log in %d ms".format(time.milliseconds - sTime)) - val errorResults = (unverifiedEntries ++ errorsPerPartition).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 - ) - } - - val allResults = localProduceResults ++ errorResults - val produceStatus = buildProducePartitionStatus(allResults) + val produceStatus = buildProducePartitionStatus(localProduceResults) - addCompletePurgatoryAction(actionQueue, allResults) - recordConversionStatsCallback(localProduceResults.map { case (k, v) => + addCompletePurgatoryAction(actionQueue, localProduceResults) + recordValidationStatsCallback(localProduceResults.map { case (k, v) => k -> v.info.recordValidationStats }) @@ -883,46 +786,100 @@ class ReplicaManager(val config: KafkaConfig, requiredAcks, delayedProduceLock, timeout, - allEntries, - allResults, + entriesPerPartition, + localProduceResults, 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]() + /** + * 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 transactionalId the transactional ID for the produce request or null if there is none. + * @param entriesPerPartition the records per partition to be appended + * @param responseCallback callback for sending the response + * @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. + */ + def handleProduceAppend(timeout: Long, + requiredAcks: Short, + internalTopicsAllowed: Boolean, + 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 + ) + } + val entriesWithoutErrorsPerPartition = entriesPerPartition.filter { case (key, _) => !errorResults.contains(key) } + + val preAppendPartitionResponses = buildProducePartitionStatus(errorResults).map { case (k, status) => k -> status.responseStatus } + + def newResponseCallback(responses: Map[TopicPartition, PartitionResponse]): Unit = { + responseCallback(preAppendPartitionResponses ++ responses) + } + + appendRecords( + timeout = timeout, + requiredAcks = requiredAcks, + internalTopicsAllowed = internalTopicsAllowed, + origin = AppendOrigin.CLIENT, + entriesPerPartition = entriesWithoutErrorsPerPartition, + responseCallback = newResponseCallback, + recordValidationStatsCallback = recordValidationStatsCallback, + requestLocal = newRequestLocal, + actionQueue = actionQueue, + verificationGuards = verificationGuards + ) + } + + 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 buildProducePartitionStatus( @@ -967,79 +924,6 @@ class ReplicaManager(val config: KafkaConfig, } } - /** - * Append messages to offsets topic, and wait for them to be replicated to other replicas; - * the callback function will be triggered either when timeout or the required acks are satisfied; - * if the callback function itself is already synchronized on some object then pass this object to avoid deadlock. - * This method should not return until the write to the local log is completed because updating offsets requires updating - * the in-memory and persisted state under a lock together. - * - * Noted that all pending delayed check operations are stored in a queue. All callers to ReplicaManager.appendRecords() - * are expected to call ActionQueue.tryCompleteActions for all affected partitions, without holding any conflicting - * locks. - * - * If appending transactional records, call maybeStartTransactionVerificationForPartition(s) and call this method in the callback. - * For example, the GroupCoordinator will pass `doCommitTxnOffsets` as the post-verification callback, and that method eventually call this. - * - * @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 entriesPerPartition the records per partition to be appended - * @param responseCallback callback for sending the response - * @param delayedProduceLock lock for the delayed actions - * @param requestLocal container for the stateful instances scoped to this request -- this must correspond to the - * thread calling this method - * @param verificationGuards the mapping from topic partition to verification guards if transaction verification is used - * @param actionQueue the action queue to use - */ - def appendForGroup( - timeout: Long, - requiredAcks: Short, - entriesPerPartition: Map[TopicPartition, MemoryRecords], - responseCallback: Map[TopicPartition, PartitionResponse] => Unit, - delayedProduceLock: Option[Lock], - requestLocal: RequestLocal, - verificationGuards: Map[TopicPartition, VerificationGuard], - actionQueue: ActionQueue = this.defaultActionQueue - ): Unit = { - if (!isValidRequiredAcks(requiredAcks)) { - sendInvalidRequiredAcksResponse(entriesPerPartition, responseCallback) - return - } - - // This should only be called on __consumer_offsets until KAFKA-15987 is complete - entriesPerPartition.keys.foreach { tp => - if (!tp.topic.equals(Topic.GROUP_METADATA_TOPIC_NAME)) - throw new IllegalArgumentException() - } - - val sTime = time.milliseconds - val localProduceResults = appendToLocalLog( - internalTopicsAllowed = true, - origin = AppendOrigin.COORDINATOR, - entriesPerPartition, - requiredAcks, - requestLocal, - verificationGuards.toMap - ) - - debug("Produce to local log in %d ms".format(time.milliseconds - sTime)) - - val allResults = localProduceResults - val produceStatus = buildProducePartitionStatus(allResults) - - addCompletePurgatoryAction(actionQueue, allResults) - - maybeAddDelayedProduce( - requiredAcks, - delayedProduceLock, - timeout, - entriesPerPartition, - allResults, - produceStatus, - responseCallback - ) - } - private def maybeAddDelayedProduce( requiredAcks: Short, delayedProduceLock: Option[Lock], @@ -1096,7 +980,7 @@ class ReplicaManager(val config: KafkaConfig, * * When the verification returns, the callback will be supplied the error if it exists or Errors.NONE. * If the verification guard exists, it will also be supplied. Otherwise the SENTINEL verification guard will be returned. - * This guard can not be used for verification and any appends that attenpt to use it will fail. + * This guard can not be used for verification and any appends that attempt to use it will fail. */ def maybeStartTransactionVerificationForPartition( topicPartition: TopicPartition, @@ -1143,7 +1027,7 @@ class ReplicaManager(val config: KafkaConfig, * is scheduled on a request handler thread. There, it should be called with that request handler thread's thread local and * not the one supplied to this method. */ - def maybeStartTransactionVerificationForPartitions( + private def maybeStartTransactionVerificationForPartitions( topicPartitionBatchInfo: Map[TopicPartition, Int], transactionalId: String, producerId: Long, @@ -1188,18 +1072,7 @@ class ReplicaManager(val config: KafkaConfig, requestLocal: RequestLocal, verificationErrors: Map[TopicPartition, Errors] ): Unit = { - // Map transaction coordinator errors to known errors for the response - val convertedErrors = verificationErrors.map { case (tp, error) => - error match { - case Errors.CONCURRENT_TRANSACTIONS | - Errors.COORDINATOR_LOAD_IN_PROGRESS | - Errors.COORDINATOR_NOT_AVAILABLE | - Errors.NOT_COORDINATOR => tp -> Errors.NOT_ENOUGH_REPLICAS - case _ => tp -> error - } - - } - callback(errors ++ convertedErrors, requestLocal, verificationGuards.toMap) + callback(errors ++ verificationErrors, requestLocal, verificationGuards.toMap) } // Wrap the callback to be handled on an arbitrary request handler thread when transaction verification is complete. @@ -1208,13 +1081,13 @@ class ReplicaManager(val config: KafkaConfig, requestLocal ) - addPartitionsToTxnManager.get.verifyTransaction( + addPartitionsToTxnManager.foreach(_.verifyTransaction( transactionalId = transactionalId, producerId = producerId, producerEpoch = producerEpoch, topicPartitions = verificationGuards.keys.toSeq, callback = verificationCallback - ) + )) } diff --git a/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala b/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala index 086ee3fae1681..b8d8afaff06ac 100644 --- a/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala @@ -209,18 +209,6 @@ object AbstractCoordinatorConcurrencyTest { override def tryCompleteActions(): Unit = watchKeys.map(producePurgatory.checkAndComplete) - override def appendForGroup(timeout: Long, - requiredAcks: Short, - entriesPerPartition: Map[TopicPartition, MemoryRecords], - responseCallback: Map[TopicPartition, PartitionResponse] => Unit, - delayedProduceLock: Option[Lock] = None, - requestLocal: RequestLocal = RequestLocal.NoCaching, - verificationGuards: Map[TopicPartition, VerificationGuard] = Map.empty, - actionQueue: ActionQueue = null): Unit = { - appendRecords(timeout, requiredAcks, true, AppendOrigin.COORDINATOR, entriesPerPartition, responseCallback, - delayedProduceLock, requestLocal = requestLocal) - } - override def appendRecords(timeout: Long, requiredAcks: Short, internalTopicsAllowed: Boolean, @@ -230,8 +218,8 @@ object AbstractCoordinatorConcurrencyTest { delayedProduceLock: Option[Lock] = None, processingStatsCallback: Map[TopicPartition, RecordValidationStats] => Unit = _ => (), requestLocal: RequestLocal = RequestLocal.NoCaching, - transactionalId: String = null, - actionQueue: ActionQueue = null): Unit = { + actionQueue: ActionQueue = null, + verificationGuards: Map[TopicPartition, VerificationGuard] = Map.empty): Unit = { if (entriesPerPartition.isEmpty) return diff --git a/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorPartitionWriterTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorPartitionWriterTest.scala index e43cc7956e731..abba3522b87de 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorPartitionWriterTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/CoordinatorPartitionWriterTest.scala @@ -27,7 +27,7 @@ import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse import org.apache.kafka.common.requests.TransactionResult import org.apache.kafka.common.utils.{MockTime, Time} import org.apache.kafka.coordinator.group.runtime.PartitionWriter -import org.apache.kafka.storage.internals.log.{LogConfig, VerificationGuard} +import org.apache.kafka.storage.internals.log.{AppendOrigin, LogConfig, VerificationGuard} import org.apache.kafka.test.TestUtils.assertFutureThrows import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows, assertTrue} import org.junit.jupiter.api.Test @@ -105,15 +105,18 @@ class CoordinatorPartitionWriterTest { val callbackCapture: ArgumentCaptor[Map[TopicPartition, PartitionResponse] => Unit] = ArgumentCaptor.forClass(classOf[Map[TopicPartition, PartitionResponse] => Unit]) - when(replicaManager.appendForGroup( + when(replicaManager.appendRecords( ArgumentMatchers.eq(0L), ArgumentMatchers.eq(1.toShort), + ArgumentMatchers.eq(true), + ArgumentMatchers.eq(AppendOrigin.COORDINATOR), recordsCapture.capture(), callbackCapture.capture(), ArgumentMatchers.any(), ArgumentMatchers.any(), + ArgumentMatchers.any(), + ArgumentMatchers.any(), ArgumentMatchers.eq(Map(tp -> VerificationGuard.SENTINEL)), - ArgumentMatchers.any() )).thenAnswer( _ => { callbackCapture.getValue.apply(Map( tp -> new PartitionResponse( @@ -179,15 +182,18 @@ class CoordinatorPartitionWriterTest { val callbackCapture: ArgumentCaptor[Map[TopicPartition, PartitionResponse] => Unit] = ArgumentCaptor.forClass(classOf[Map[TopicPartition, PartitionResponse] => Unit]) - when(replicaManager.appendForGroup( + when(replicaManager.appendRecords( ArgumentMatchers.eq(0L), ArgumentMatchers.eq(1.toShort), + ArgumentMatchers.eq(true), + ArgumentMatchers.eq(AppendOrigin.COORDINATOR), recordsCapture.capture(), callbackCapture.capture(), ArgumentMatchers.any(), ArgumentMatchers.any(), + ArgumentMatchers.any(), + ArgumentMatchers.any(), ArgumentMatchers.eq(Map(tp -> verificationGuard)), - ArgumentMatchers.any() )).thenAnswer(_ => { callbackCapture.getValue.apply(Map( tp -> new PartitionResponse( @@ -258,15 +264,18 @@ class CoordinatorPartitionWriterTest { val callbackCapture: ArgumentCaptor[Map[TopicPartition, PartitionResponse] => Unit] = ArgumentCaptor.forClass(classOf[Map[TopicPartition, PartitionResponse] => Unit]) - when(replicaManager.appendForGroup( + when(replicaManager.appendRecords( ArgumentMatchers.eq(0L), ArgumentMatchers.eq(1.toShort), + ArgumentMatchers.eq(true), + ArgumentMatchers.eq(AppendOrigin.COORDINATOR), recordsCapture.capture(), callbackCapture.capture(), ArgumentMatchers.any(), ArgumentMatchers.any(), + ArgumentMatchers.any(), + ArgumentMatchers.any(), ArgumentMatchers.eq(Map(tp -> VerificationGuard.SENTINEL)), - ArgumentMatchers.any() )).thenAnswer(_ => { callbackCapture.getValue.apply(Map( tp -> new PartitionResponse( @@ -380,15 +389,18 @@ class CoordinatorPartitionWriterTest { val callbackCapture: ArgumentCaptor[Map[TopicPartition, PartitionResponse] => Unit] = ArgumentCaptor.forClass(classOf[Map[TopicPartition, PartitionResponse] => Unit]) - when(replicaManager.appendForGroup( + when(replicaManager.appendRecords( ArgumentMatchers.eq(0L), ArgumentMatchers.eq(1.toShort), + ArgumentMatchers.eq(true), + ArgumentMatchers.eq(AppendOrigin.COORDINATOR), recordsCapture.capture(), callbackCapture.capture(), ArgumentMatchers.any(), ArgumentMatchers.any(), + ArgumentMatchers.any(), + ArgumentMatchers.any(), ArgumentMatchers.eq(Map(tp -> VerificationGuard.SENTINEL)), - ArgumentMatchers.any() )).thenAnswer(_ => { callbackCapture.getValue.apply(Map( tp -> new PartitionResponse(Errors.NOT_LEADER_OR_FOLLOWER) diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala index 2fdfde39c1ee3..c674dfa0a6cdf 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala @@ -39,7 +39,7 @@ import org.apache.kafka.common.message.LeaveGroupRequestData.MemberIdentity import org.apache.kafka.coordinator.group.OffsetConfig import org.apache.kafka.server.util.timer.MockTimer import org.apache.kafka.server.util.{KafkaScheduler, MockTime} -import org.apache.kafka.storage.internals.log.VerificationGuard +import org.apache.kafka.storage.internals.log.{AppendOrigin, VerificationGuard} import org.junit.jupiter.api.Assertions._ import org.junit.jupiter.api.{AfterEach, BeforeEach, Test} import org.junit.jupiter.params.ParameterizedTest @@ -3897,14 +3897,17 @@ class GroupCoordinatorTest { val capturedArgument: ArgumentCaptor[scala.collection.Map[TopicPartition, PartitionResponse] => Unit] = ArgumentCaptor.forClass(classOf[scala.collection.Map[TopicPartition, PartitionResponse] => Unit]) - when(replicaManager.appendForGroup(anyLong, + when(replicaManager.appendRecords(anyLong, anyShort(), + internalTopicsAllowed = ArgumentMatchers.eq(true), + origin = ArgumentMatchers.eq(AppendOrigin.COORDINATOR), any[Map[TopicPartition, MemoryRecords]], capturedArgument.capture(), any[Option[ReentrantLock]], + any(), any(classOf[RequestLocal]), - any[Map[TopicPartition, VerificationGuard]], - any[ActionQueue] + any[ActionQueue], + any[Map[TopicPartition, VerificationGuard]] )).thenAnswer(_ => { capturedArgument.getValue.apply( Map(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, groupPartitionId) -> @@ -3930,14 +3933,17 @@ class GroupCoordinatorTest { val capturedArgument: ArgumentCaptor[scala.collection.Map[TopicPartition, PartitionResponse] => Unit] = ArgumentCaptor.forClass(classOf[scala.collection.Map[TopicPartition, PartitionResponse] => Unit]) - when(replicaManager.appendForGroup(anyLong, + when(replicaManager.appendRecords(anyLong, anyShort(), + internalTopicsAllowed = ArgumentMatchers.eq(true), + origin = ArgumentMatchers.eq(AppendOrigin.COORDINATOR), any[Map[TopicPartition, MemoryRecords]], capturedArgument.capture(), any[Option[ReentrantLock]], + any(), any(classOf[RequestLocal]), - any[Map[TopicPartition, VerificationGuard]], - any[ActionQueue] + any[ActionQueue], + any[Map[TopicPartition, VerificationGuard]] )).thenAnswer(_ => { capturedArgument.getValue.apply( Map(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, groupPartitionId) -> @@ -4074,14 +4080,17 @@ class GroupCoordinatorTest { val capturedArgument: ArgumentCaptor[scala.collection.Map[TopicPartition, PartitionResponse] => Unit] = ArgumentCaptor.forClass(classOf[scala.collection.Map[TopicPartition, PartitionResponse] => Unit]) - when(replicaManager.appendForGroup(anyLong, + when(replicaManager.appendRecords(anyLong, anyShort(), + internalTopicsAllowed = ArgumentMatchers.eq(true), + origin = ArgumentMatchers.eq(AppendOrigin.COORDINATOR), any[Map[TopicPartition, MemoryRecords]], capturedArgument.capture(), any[Option[ReentrantLock]], + any(), any(classOf[RequestLocal]), - any[Map[TopicPartition, VerificationGuard]], - any[ActionQueue] + any[ActionQueue], + any[Map[TopicPartition, VerificationGuard]] )).thenAnswer(_ => { capturedArgument.getValue.apply( Map(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, groupPartitionId) -> @@ -4117,14 +4126,17 @@ class GroupCoordinatorTest { ArgumentMatchers.eq(producerId), ArgumentMatchers.eq(producerEpoch), any(), any(), postVerificationCallback.capture())).thenAnswer( _ => postVerificationCallback.getValue()(verificationError, RequestLocal.NoCaching, VerificationGuard.SENTINEL) ) - when(replicaManager.appendForGroup(anyLong, + when(replicaManager.appendRecords(anyLong, anyShort(), + internalTopicsAllowed = ArgumentMatchers.eq(true), + origin = ArgumentMatchers.eq(AppendOrigin.COORDINATOR), any[Map[TopicPartition, MemoryRecords]], capturedArgument.capture(), any[Option[ReentrantLock]], + any(), any(classOf[RequestLocal]), - any[Map[TopicPartition, VerificationGuard]], - any[ActionQueue] + any[ActionQueue], + any[Map[TopicPartition, VerificationGuard]] )).thenAnswer(_ => { capturedArgument.getValue.apply( Map(offsetTopicPartition -> diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala index 5fcbad9a2a521..10b2ebe896505 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala @@ -1176,13 +1176,16 @@ class GroupMetadataManagerTest { groupMetadataManager.storeGroup(group, Map.empty, callback) assertEquals(Some(expectedError), maybeError) - verify(replicaManager).appendForGroup(anyLong(), + verify(replicaManager).appendRecords(anyLong(), anyShort(), any(), any(), + any(), + any(), any[Option[ReentrantLock]], any(), any(), + any(), any()) verify(replicaManager).getMagic(any()) } @@ -1211,13 +1214,16 @@ class GroupMetadataManagerTest { groupMetadataManager.storeGroup(group, Map(memberId -> Array[Byte]()), callback) assertEquals(Some(Errors.NONE), maybeError) - verify(replicaManager).appendForGroup(anyLong(), + verify(replicaManager).appendRecords(anyLong(), anyShort(), any(), any(), + any(), + any(), any[Option[ReentrantLock]], any(), any(), + any(), any()) verify(replicaManager).getMagic(any()) } @@ -1285,13 +1291,16 @@ class GroupMetadataManagerTest { assertEquals(Errors.NONE, partitionResponse.error) assertEquals(offset, partitionResponse.offset) - verify(replicaManager).appendForGroup(anyLong(), + verify(replicaManager).appendRecords(anyLong(), anyShort(), any(), any(), + any(), + any(), any[Option[ReentrantLock]], any(), any(), + any(), any()) // Will update sensor after commit assertEquals(1, TestUtils.totalMetricValue(metrics, "offset-commit-count")) @@ -1326,14 +1335,17 @@ class GroupMetadataManagerTest { assertTrue(group.hasOffsets) assertTrue(group.allOffsets.isEmpty) - verify(replicaManager).appendForGroup(anyLong(), + verify(replicaManager).appendRecords(anyLong(), anyShort(), + any(), + any(), any[Map[TopicPartition, MemoryRecords]], capturedResponseCallback.capture(), any[Option[ReentrantLock]], any(), - ArgumentMatchers.eq(Map(offsetTopicPartition -> verificationGuard)), - any()) + any(), + any(), + ArgumentMatchers.eq(Map(offsetTopicPartition -> verificationGuard))) verify(replicaManager).getMagic(any()) capturedResponseCallback.getValue.apply(Map(groupTopicPartition -> new PartitionResponse(Errors.NONE, 0L, RecordBatch.NO_TIMESTAMP, 0L))) @@ -1385,14 +1397,17 @@ class GroupMetadataManagerTest { assertFalse(group.hasOffsets) assertTrue(group.allOffsets.isEmpty) - verify(replicaManager).appendForGroup(anyLong(), + verify(replicaManager).appendRecords(anyLong(), anyShort(), + any(), + any(), any[Map[TopicPartition, MemoryRecords]], any(), any[Option[ReentrantLock]], any(), - ArgumentMatchers.eq(Map(offsetTopicPartition -> verificationGuard)), - any()) + any(), + any(), + ArgumentMatchers.eq(Map(offsetTopicPartition -> verificationGuard))) verify(replicaManager).getMagic(any()) } @@ -1434,14 +1449,17 @@ class GroupMetadataManagerTest { assertFalse(group.hasOffsets) assertTrue(group.allOffsets.isEmpty) - verify(replicaManager).appendForGroup(anyLong(), + verify(replicaManager).appendRecords(anyLong(), anyShort(), + any(), + any(), any[Map[TopicPartition, MemoryRecords]], any(), any[Option[ReentrantLock]], any(), - ArgumentMatchers.eq(Map(offsetTopicPartition -> verificationGuard)), - any()) + any(), + any(), + ArgumentMatchers.eq(Map(offsetTopicPartition -> verificationGuard))) verify(replicaManager).getMagic(any()) } @@ -1586,13 +1604,16 @@ class GroupMetadataManagerTest { cachedOffsets.get(topicIdPartitionFailed.topicPartition).map(_.offset) ) - verify(replicaManager).appendForGroup(anyLong(), + verify(replicaManager).appendRecords(anyLong(), anyShort(), + any(), + any(), any[Map[TopicPartition, MemoryRecords]], any(), any[Option[ReentrantLock]], any(), any(), + any(), any()) verify(replicaManager).getMagic(any()) assertEquals(1, TestUtils.totalMetricValue(metrics, "offset-commit-count")) @@ -1692,13 +1713,16 @@ class GroupMetadataManagerTest { assertEquals(Some(OffsetFetchResponse.INVALID_OFFSET), cachedOffsets.get(topicIdPartition1.topicPartition).map(_.offset)) assertEquals(Some(offset), cachedOffsets.get(topicIdPartition2.topicPartition).map(_.offset)) - verify(replicaManager).appendForGroup(anyLong(), + verify(replicaManager).appendRecords(anyLong(), anyShort(), any(), any(), + any(), + any(), any[Option[ReentrantLock]], any(), any(), + any(), any()) verify(replicaManager, times(2)).getMagic(any()) } @@ -2800,13 +2824,16 @@ class GroupMetadataManagerTest { private def verifyAppendAndCaptureCallback(): ArgumentCaptor[Map[TopicPartition, PartitionResponse] => Unit] = { val capturedArgument: ArgumentCaptor[Map[TopicPartition, PartitionResponse] => Unit] = ArgumentCaptor.forClass(classOf[Map[TopicPartition, PartitionResponse] => Unit]) - verify(replicaManager).appendForGroup(anyLong(), + verify(replicaManager).appendRecords(anyLong(), anyShort(), + any(), + any(), any[Map[TopicPartition, MemoryRecords]], capturedArgument.capture(), any[Option[ReentrantLock]], any(), any(), + any(), any()) capturedArgument } @@ -2814,13 +2841,16 @@ class GroupMetadataManagerTest { private def expectAppendMessage(error: Errors): ArgumentCaptor[Map[TopicPartition, MemoryRecords]] = { val capturedCallback: ArgumentCaptor[Map[TopicPartition, PartitionResponse] => Unit] = ArgumentCaptor.forClass(classOf[Map[TopicPartition, PartitionResponse] => Unit]) val capturedRecords: ArgumentCaptor[Map[TopicPartition, MemoryRecords]] = ArgumentCaptor.forClass(classOf[Map[TopicPartition, MemoryRecords]]) - when(replicaManager.appendForGroup(anyLong(), + when(replicaManager.appendRecords(anyLong(), anyShort(), + any(), + any(), capturedRecords.capture(), capturedCallback.capture(), any[Option[ReentrantLock]], any(), any(), + any(), any() )).thenAnswer(_ => { capturedCallback.getValue.apply( diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala index 721157279e8df..a0a7fd79b1d41 100644 --- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala @@ -2486,14 +2486,12 @@ class KafkaApisTest extends Logging { .build(version.toShort) val request = buildRequest(produceRequest) - when(replicaManager.appendRecords(anyLong, + when(replicaManager.handleProduceAppend(anyLong, anyShort, ArgumentMatchers.eq(false), - ArgumentMatchers.eq(AppendOrigin.CLIENT), - any(), - responseCallback.capture(), any(), any(), + responseCallback.capture(), any(), any(), any() @@ -2549,14 +2547,12 @@ class KafkaApisTest extends Logging { .build(version.toShort) val request = buildRequest(produceRequest) - when(replicaManager.appendRecords(anyLong, + when(replicaManager.handleProduceAppend(anyLong, anyShort, ArgumentMatchers.eq(false), - ArgumentMatchers.eq(AppendOrigin.CLIENT), - any(), - responseCallback.capture(), any(), any(), + responseCallback.capture(), any(), any(), any()) @@ -2615,14 +2611,12 @@ class KafkaApisTest extends Logging { .build(version.toShort) val request = buildRequest(produceRequest) - when(replicaManager.appendRecords(anyLong, + when(replicaManager.handleProduceAppend(anyLong, anyShort, ArgumentMatchers.eq(false), - ArgumentMatchers.eq(AppendOrigin.CLIENT), - any(), - responseCallback.capture(), any(), any(), + responseCallback.capture(), any(), any(), any()) @@ -2680,14 +2674,12 @@ class KafkaApisTest extends Logging { .build(version.toShort) val request = buildRequest(produceRequest) - when(replicaManager.appendRecords(anyLong, + when(replicaManager.handleProduceAppend(anyLong, anyShort, ArgumentMatchers.eq(false), - ArgumentMatchers.eq(AppendOrigin.CLIENT), - any(), - responseCallback.capture(), any(), any(), + responseCallback.capture(), any(), any(), any()) @@ -2729,8 +2721,6 @@ class KafkaApisTest extends Logging { reset(replicaManager, clientQuotaManager, clientRequestQuotaManager, requestChannel, txnCoordinator) - val responseCallback: ArgumentCaptor[Map[TopicPartition, PartitionResponse] => Unit] = ArgumentCaptor.forClass(classOf[Map[TopicPartition, PartitionResponse] => Unit]) - val tp = new TopicPartition("topic", 0) val produceRequest = ProduceRequest.forCurrentMagic(new ProduceRequestData() @@ -2751,16 +2741,14 @@ class KafkaApisTest extends Logging { try { kafkaApis.handleProduceRequest(request, RequestLocal.withThreadConfinedCaching) - verify(replicaManager).appendRecords(anyLong, + verify(replicaManager).handleProduceAppend(anyLong, anyShort, ArgumentMatchers.eq(false), - ArgumentMatchers.eq(AppendOrigin.CLIENT), + ArgumentMatchers.eq(transactionalId), any(), - responseCallback.capture(), any(), any(), any(), - ArgumentMatchers.eq(transactionalId), any()) } finally { kafkaApis.close() diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala index 9201db0765e02..499ae8a13a06b 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala @@ -670,7 +670,7 @@ class ReplicaManagerTest { val sequence = 9 val records = MemoryRecords.withTransactionalRecords(CompressionType.NONE, producerId, epoch, sequence, new SimpleRecord(time.milliseconds(), s"message $sequence".getBytes)) - appendRecords(replicaManager, new TopicPartition(topic, 0), records, transactionalId = transactionalId).onFire { response => + handleProduceAppend(replicaManager, new TopicPartition(topic, 0), records, transactionalId = transactionalId).onFire { response => assertEquals(Errors.NONE, response.error) } assertLateTransactionCount(Some(0)) @@ -734,7 +734,7 @@ class ReplicaManagerTest { for (sequence <- 0 until numRecords) { val records = MemoryRecords.withTransactionalRecords(CompressionType.NONE, producerId, epoch, sequence, new SimpleRecord(s"message $sequence".getBytes)) - appendRecords(replicaManager, new TopicPartition(topic, 0), records, transactionalId = transactionalId).onFire { response => + handleProduceAppend(replicaManager, new TopicPartition(topic, 0), records, transactionalId = transactionalId).onFire { response => assertEquals(Errors.NONE, response.error) } } @@ -855,7 +855,7 @@ class ReplicaManagerTest { for (sequence <- 0 until numRecords) { val records = MemoryRecords.withTransactionalRecords(CompressionType.NONE, producerId, epoch, sequence, new SimpleRecord(s"message $sequence".getBytes)) - appendRecords(replicaManager, new TopicPartition(topic, 0), records, transactionalId = transactionalId).onFire { response => + handleProduceAppend(replicaManager, new TopicPartition(topic, 0), records, transactionalId = transactionalId).onFire { response => assertEquals(Errors.NONE, response.error) } } @@ -2178,7 +2178,7 @@ class ReplicaManagerTest { // If we supply no transactional ID and idempotent records, we do not verify. val idempotentRecords = MemoryRecords.withIdempotentRecords(CompressionType.NONE, producerId, producerEpoch, sequence, new SimpleRecord("message".getBytes)) - appendRecords(replicaManager, tp0, idempotentRecords) + handleProduceAppend(replicaManager, tp0, idempotentRecords, transactionalId = null) verify(addPartitionsToTxnManager, times(0)).verifyTransaction(any(), any(), any(), any(), any[AddPartitionsToTxnManager.AppendCallback]()) assertEquals(VerificationGuard.SENTINEL, getVerificationGuard(replicaManager, tp0, producerId)) @@ -2188,7 +2188,7 @@ class ReplicaManagerTest { val idempotentRecords2 = MemoryRecords.withIdempotentRecords(CompressionType.NONE, producerId, producerEpoch, sequence, new SimpleRecord("message".getBytes)) - appendRecordsToMultipleTopics(replicaManager, Map(tp0 -> transactionalRecords, tp1 -> idempotentRecords2), transactionalId) + handleProduceAppendToMultipleTopics(replicaManager, Map(tp0 -> transactionalRecords, tp1 -> idempotentRecords2), transactionalId) verify(addPartitionsToTxnManager, times(1)).verifyTransaction( ArgumentMatchers.eq(transactionalId), ArgumentMatchers.eq(producerId), @@ -2223,7 +2223,7 @@ class ReplicaManagerTest { new SimpleRecord("message".getBytes)) // We should add these partitions to the manager to verify. - val result = appendRecords(replicaManager, tp0, transactionalRecords, origin = appendOrigin, transactionalId = transactionalId) + val result = handleProduceAppend(replicaManager, tp0, transactionalRecords, origin = appendOrigin, transactionalId = transactionalId) val appendCallback = ArgumentCaptor.forClass(classOf[AddPartitionsToTxnManager.AppendCallback]) verify(addPartitionsToTxnManager, times(1)).verifyTransaction( ArgumentMatchers.eq(transactionalId), @@ -2242,7 +2242,7 @@ class ReplicaManagerTest { assertEquals(verificationGuard, getVerificationGuard(replicaManager, tp0, producerId)) // This time verification is successful. - appendRecords(replicaManager, tp0, transactionalRecords, origin = appendOrigin, transactionalId = transactionalId) + handleProduceAppend(replicaManager, tp0, transactionalRecords, origin = appendOrigin, transactionalId = transactionalId) val appendCallback2 = ArgumentCaptor.forClass(classOf[AddPartitionsToTxnManager.AppendCallback]) verify(addPartitionsToTxnManager, times(2)).verifyTransaction( ArgumentMatchers.eq(transactionalId), @@ -2281,7 +2281,7 @@ class ReplicaManagerTest { new SimpleRecord("message".getBytes)) // We should add these partitions to the manager to verify. - val result = appendRecords(replicaManager, tp0, transactionalRecords, transactionalId = transactionalId) + val result = handleProduceAppend(replicaManager, tp0, transactionalRecords, transactionalId = transactionalId) val appendCallback = ArgumentCaptor.forClass(classOf[AddPartitionsToTxnManager.AppendCallback]) verify(addPartitionsToTxnManager, times(1)).verifyTransaction( ArgumentMatchers.eq(transactionalId), @@ -2303,7 +2303,7 @@ class ReplicaManagerTest { val transactionalRecords2 = MemoryRecords.withTransactionalRecords(CompressionType.NONE, producerId, producerEpoch, sequence + 1, new SimpleRecord("message".getBytes)) - val result2 = appendRecords(replicaManager, tp0, transactionalRecords2, transactionalId = transactionalId) + val result2 = handleProduceAppend(replicaManager, tp0, transactionalRecords2, transactionalId = transactionalId) val appendCallback2 = ArgumentCaptor.forClass(classOf[AddPartitionsToTxnManager.AppendCallback]) verify(addPartitionsToTxnManager, times(2)).verifyTransaction( ArgumentMatchers.eq(transactionalId), @@ -2346,7 +2346,7 @@ class ReplicaManagerTest { val transactionalRecords = MemoryRecords.withTransactionalRecords(CompressionType.NONE, producerId, producerEpoch, sequence, new SimpleRecord(s"message $sequence".getBytes)) - appendRecordsToMultipleTopics(replicaManager, Map(tp0 -> transactionalRecords, tp1 -> transactionalRecords), transactionalId).onFire { responses => + handleProduceAppendToMultipleTopics(replicaManager, Map(tp0 -> transactionalRecords, tp1 -> transactionalRecords), transactionalId).onFire { responses => responses.foreach { entry => assertEquals(Errors.NONE, entry._2.error) } @@ -2386,7 +2386,7 @@ class ReplicaManagerTest { new SimpleRecord(s"message $sequence".getBytes))) assertThrows(classOf[InvalidPidMappingException], - () => appendRecordsToMultipleTopics(replicaManager, transactionalRecords, transactionalId = transactionalId)) + () => handleProduceAppendToMultipleTopics(replicaManager, transactionalRecords, transactionalId = transactionalId)) // We should not add these partitions to the manager to verify. verify(addPartitionsToTxnManager, times(0)).verifyTransaction(any(), any(), any(), any(), any()) } finally { @@ -2409,7 +2409,7 @@ class ReplicaManagerTest { new SimpleRecord("message".getBytes)) // We should not add these partitions to the manager to verify, but instead throw an error. - appendRecords(replicaManager, tp0, transactionalRecords, transactionalId = transactionalId).onFire { response => + handleProduceAppend(replicaManager, tp0, transactionalRecords, transactionalId = transactionalId).onFire { response => assertEquals(Errors.NOT_LEADER_OR_FOLLOWER, response.error) } verify(addPartitionsToTxnManager, times(0)).verifyTransaction(any(), any(), any(), any(), any()) @@ -2440,7 +2440,9 @@ class ReplicaManagerTest { val transactionalRecords = MemoryRecords.withTransactionalRecords(CompressionType.NONE, producerId, producerEpoch, sequence, new SimpleRecord(s"message $sequence".getBytes)) - appendRecords(replicaManager, tp, transactionalRecords, transactionalId = transactionalId) + handleProduceAppend(replicaManager, tp, transactionalRecords, transactionalId = transactionalId).onFire { response => + assertEquals(Errors.NONE, response.error) + } assertEquals(VerificationGuard.SENTINEL, getVerificationGuard(replicaManager, tp, producerId)) // We should not add these partitions to the manager to verify. @@ -2457,7 +2459,7 @@ class ReplicaManagerTest { val moreTransactionalRecords = MemoryRecords.withTransactionalRecords(CompressionType.NONE, producerId, producerEpoch, sequence + 1, new SimpleRecord("message".getBytes)) - appendRecords(replicaManager, tp, moreTransactionalRecords, transactionalId = transactionalId) + handleProduceAppend(replicaManager, tp, moreTransactionalRecords, transactionalId = transactionalId) verify(addPartitionsToTxnManager, times(0)).verifyTransaction(any(), any(), any(), any(), any()) assertEquals(VerificationGuard.SENTINEL, getVerificationGuard(replicaManager, tp, producerId)) assertTrue(replicaManager.localLog(tp).get.hasOngoingTransaction(producerId)) @@ -2485,7 +2487,7 @@ class ReplicaManagerTest { new SimpleRecord("message".getBytes)) // We should add these partitions to the manager to verify. - val result = appendRecords(replicaManager, tp0, transactionalRecords, transactionalId = transactionalId) + val result = handleProduceAppend(replicaManager, tp0, transactionalRecords, transactionalId = transactionalId) val appendCallback = ArgumentCaptor.forClass(classOf[AddPartitionsToTxnManager.AppendCallback]) verify(addPartitionsToTxnManager, times(1)).verifyTransaction( ArgumentMatchers.eq(transactionalId), @@ -2511,7 +2513,7 @@ class ReplicaManagerTest { assertEquals(verificationGuard, getVerificationGuard(replicaManager, tp0, producerId)) // This time we do not verify - appendRecords(replicaManager, tp0, transactionalRecords, transactionalId = transactionalId) + handleProduceAppend(replicaManager, tp0, transactionalRecords, transactionalId = transactionalId) verify(addPartitionsToTxnManager, times(1)).verifyTransaction(any(), any(), any(), any(), any()) assertEquals(VerificationGuard.SENTINEL, getVerificationGuard(replicaManager, tp0, producerId)) assertTrue(replicaManager.localLog(tp0).get.hasOngoingTransaction(producerId)) @@ -2540,7 +2542,7 @@ class ReplicaManagerTest { // Start verification and return the coordinator related errors. val expectedMessage = s"Unable to verify the partition has been added to the transaction. Underlying error: ${error.toString}" - val result = appendRecords(replicaManager, tp0, transactionalRecords, transactionalId = transactionalId) + val result = handleProduceAppend(replicaManager, tp0, transactionalRecords, transactionalId = transactionalId) val appendCallback = ArgumentCaptor.forClass(classOf[AddPartitionsToTxnManager.AppendCallback]) verify(addPartitionsToTxnManager, times(1)).verifyTransaction( ArgumentMatchers.eq(transactionalId), @@ -2560,41 +2562,6 @@ class ReplicaManagerTest { } } - @ParameterizedTest - @EnumSource(value = classOf[Errors], names = Array("NOT_COORDINATOR", "CONCURRENT_TRANSACTIONS", "COORDINATOR_LOAD_IN_PROGRESS", "COORDINATOR_NOT_AVAILABLE")) - def testMaybeVerificationErrorConversions(error: Errors): Unit = { - val tp0 = new TopicPartition(topic, 0) - val transactionalId = "txn-id" - val producerId = 24L - val producerEpoch = 0.toShort - val addPartitionsToTxnManager = mock(classOf[AddPartitionsToTxnManager]) - - val replicaManager = setUpReplicaManagerWithMockedAddPartitionsToTxnManager(addPartitionsToTxnManager, List(tp0)) - try { - replicaManager.becomeLeaderOrFollower(1, - makeLeaderAndIsrRequest(topicIds(tp0.topic), tp0, Seq(0, 1), LeaderAndIsr(1, List(0, 1))), - (_, _) => ()) - - // Start verification and return the coordinator related errors. - val result = maybeStartTransactionVerificationForPartition(replicaManager, tp0, transactionalId, producerId, producerEpoch) - val appendCallback = ArgumentCaptor.forClass(classOf[AddPartitionsToTxnManager.AppendCallback]) - verify(addPartitionsToTxnManager, times(1)).verifyTransaction( - ArgumentMatchers.eq(transactionalId), - ArgumentMatchers.eq(producerId), - ArgumentMatchers.eq(producerEpoch), - ArgumentMatchers.eq(Seq(tp0)), - appendCallback.capture() - ) - - // Confirm we did not write to the log and instead returned the converted error with the correct error message. - val callback: AddPartitionsToTxnManager.AppendCallback = appendCallback.getValue() - callback(Map(tp0 -> error).toMap) - assertEquals(Errors.NOT_ENOUGH_REPLICAS, result.assertFired.left.getOrElse(Errors.NONE)) - } finally { - replicaManager.shutdown(checkpointHW = false) - } - } - @Test def testPreVerificationError(): Unit = { val tp0 = new TopicPartition(topic, 0) @@ -3025,8 +2992,7 @@ class ReplicaManagerTest { partition: TopicPartition, records: MemoryRecords, origin: AppendOrigin = AppendOrigin.CLIENT, - requiredAcks: Short = -1, - transactionalId: String = null): CallbackResult[PartitionResponse] = { + requiredAcks: Short = -1): CallbackResult[PartitionResponse] = { val result = new CallbackResult[PartitionResponse]() def appendCallback(responses: Map[TopicPartition, PartitionResponse]): Unit = { val response = responses.get(partition) @@ -3041,31 +3007,56 @@ class ReplicaManagerTest { origin = origin, entriesPerPartition = Map(partition -> records), responseCallback = appendCallback, - transactionalId = transactionalId, ) result } - private def appendRecordsToMultipleTopics(replicaManager: ReplicaManager, - entriesToAppend: Map[TopicPartition, MemoryRecords], - transactionalId: String, - origin: AppendOrigin = AppendOrigin.CLIENT, - requiredAcks: Short = -1): CallbackResult[Map[TopicPartition, PartitionResponse]] = { + private def handleProduceAppendToMultipleTopics(replicaManager: ReplicaManager, + entriesToAppend: Map[TopicPartition, MemoryRecords], + transactionalId: String, + origin: AppendOrigin = AppendOrigin.CLIENT, + requiredAcks: Short = -1): CallbackResult[Map[TopicPartition, PartitionResponse]] = { val result = new CallbackResult[Map[TopicPartition, PartitionResponse]]() def appendCallback(responses: Map[TopicPartition, PartitionResponse]): Unit = { responses.foreach( response => assertTrue(responses.get(response._1).isDefined)) result.fire(responses) } - replicaManager.appendRecords( + replicaManager.handleProduceAppend( timeout = 1000, requiredAcks = requiredAcks, internalTopicsAllowed = false, - origin = origin, + transactionalId = transactionalId, entriesPerPartition = entriesToAppend, responseCallback = appendCallback, + ) + + result + } + + private def handleProduceAppend(replicaManager: ReplicaManager, + partition: TopicPartition, + records: MemoryRecords, + origin: AppendOrigin = AppendOrigin.CLIENT, + requiredAcks: Short = -1, + transactionalId: String): CallbackResult[PartitionResponse] = { + val result = new CallbackResult[PartitionResponse]() + + def appendCallback(responses: Map[TopicPartition, PartitionResponse]): Unit = { + val response = responses.get(partition) + assertTrue(response.isDefined) + result.fire(response.get) + } + + val entriesPerPartition = Map(partition -> records) + replicaManager.handleProduceAppend( + timeout = 1000, + requiredAcks = requiredAcks, + internalTopicsAllowed = false, transactionalId = transactionalId, + entriesPerPartition = entriesPerPartition, + responseCallback = appendCallback, ) result