KAFKA-15987: Refactor ReplicaManager code for transaction verification

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,

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,

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)
     }
   }

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

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))
         }
       }
     }