KAFKA-18884: Move TransactionMetadata to transaction-coordinator module

core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala

@@ -21,11 +21,12 @@ import org.apache.kafka.common.compress.Compression
 import org.apache.kafka.common.protocol.{ByteBufferAccessor, MessageUtil}
 import org.apache.kafka.common.record.RecordBatch
 import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.coordinator.transaction.{TransactionState, TxnTransitMetadata}
+import org.apache.kafka.coordinator.transaction.{TransactionMetadata, TransactionState, TxnTransitMetadata}
 import org.apache.kafka.coordinator.transaction.generated.{CoordinatorRecordType, TransactionLogKey, TransactionLogValue}
 import org.apache.kafka.server.common.TransactionVersion
 
-import scala.collection.mutable
+import java.util
+
 import scala.jdk.CollectionConverters._
 
 /**
@@ -115,26 +116,26 @@ object TransactionLog {
       if (version >= TransactionLogValue.LOWEST_SUPPORTED_VERSION && version <= TransactionLogValue.HIGHEST_SUPPORTED_VERSION) {
         val value = new TransactionLogValue(new ByteBufferAccessor(buffer), version)
         val transactionMetadata = new TransactionMetadata(
-          transactionalId = transactionalId,
-          producerId = value.producerId,
-          prevProducerId = value.previousProducerId,
-          nextProducerId = value.nextProducerId,
-          producerEpoch = value.producerEpoch,
-          lastProducerEpoch = RecordBatch.NO_PRODUCER_EPOCH,
-          txnTimeoutMs = value.transactionTimeoutMs,
-          state = TransactionState.fromId(value.transactionStatus),
-          topicPartitions = mutable.Set.empty[TopicPartition],
-          txnStartTimestamp = value.transactionStartTimestampMs,
-          txnLastUpdateTimestamp = value.transactionLastUpdateTimestampMs,
-          clientTransactionVersion = TransactionVersion.fromFeatureLevel(value.clientTransactionVersion))
+          transactionalId,
+          value.producerId,
+          value.previousProducerId,
+          value.nextProducerId,
+          value.producerEpoch,
+          RecordBatch.NO_PRODUCER_EPOCH,
+          value.transactionTimeoutMs,
+          TransactionState.fromId(value.transactionStatus),
+          util.Set.of(),
+          value.transactionStartTimestampMs,
+          value.transactionLastUpdateTimestampMs,
+          TransactionVersion.fromFeatureLevel(value.clientTransactionVersion))
 
         if (!transactionMetadata.state.equals(TransactionState.EMPTY))
-          value.transactionPartitions.forEach(partitionsSchema =>
+          value.transactionPartitions.forEach(partitionsSchema => {
             transactionMetadata.addPartitions(partitionsSchema.partitionIds
-              .asScala
-              .map(partitionId => new TopicPartition(partitionsSchema.topic, partitionId))
-              .toSet)
-          )
+              .stream
+              .map(partitionId => new TopicPartition(partitionsSchema.topic, partitionId.intValue()))
+              .toList)
+          })
         Some(transactionMetadata)
       } else throw new IllegalStateException(s"Unknown version $version from the transaction log message value")
     }

core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala

@@ -32,7 +32,7 @@ import org.apache.kafka.common.requests.{TransactionResult, WriteTxnMarkersReque
 import org.apache.kafka.common.security.JaasContext
 import org.apache.kafka.common.utils.{LogContext, Time}
 import org.apache.kafka.common.{Node, Reconfigurable, TopicPartition}
-import org.apache.kafka.coordinator.transaction.TxnTransitMetadata
+import org.apache.kafka.coordinator.transaction.{TransactionMetadata, TxnTransitMetadata}
 import org.apache.kafka.metadata.MetadataCache
 import org.apache.kafka.server.common.RequestLocal
 import org.apache.kafka.server.metrics.KafkaMetricsGroup
@@ -326,16 +326,16 @@ class TransactionMarkerChannelManager(
       info(s"Replaced an existing pending complete txn $prev with $pendingCompleteTxn while adding markers to send.")
     }
     addTxnMarkersToBrokerQueue(txnMetadata.producerId,
-      txnMetadata.producerEpoch, txnResult, pendingCompleteTxn, txnMetadata.topicPartitions.toSet)
+      txnMetadata.producerEpoch, txnResult, pendingCompleteTxn, txnMetadata.topicPartitions.asScala.toSet)
     maybeWriteTxnCompletion(transactionalId)
   }
 
   def numTxnsWithPendingMarkers: Int = transactionsWithPendingMarkers.size
 
   private def hasPendingMarkersToWrite(txnMetadata: TransactionMetadata): Boolean = {
-    txnMetadata.inLock {
-      txnMetadata.topicPartitions.nonEmpty
-    }
+    txnMetadata.inLock(() =>
+      !txnMetadata.topicPartitions.isEmpty
+    )
   }
 
   def maybeWriteTxnCompletion(transactionalId: String): Unit = {
@@ -422,9 +422,9 @@ class TransactionMarkerChannelManager(
 
                 val txnMetadata = epochAndMetadata.transactionMetadata
 
-                txnMetadata.inLock {
+                txnMetadata.inLock(() =>
                   topicPartitions.foreach(txnMetadata.removePartition)
-                }
+                )
 
                 maybeWriteTxnCompletion(transactionalId)
               }

core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerRequestCompletionHandler.scala

@@ -131,7 +131,7 @@ class TransactionMarkerRequestCompletionHandler(brokerId: Int,
               txnMarkerChannelManager.removeMarkersForTxn(pendingCompleteTxn)
               abortSending = true
             } else {
-              txnMetadata.inLock {
+              txnMetadata.inLock(() => {
                 for ((topicPartition, error) <- errors.asScala) {
                   error match {
                     case Errors.NONE =>
@@ -178,7 +178,7 @@ class TransactionMarkerRequestCompletionHandler(brokerId: Int,
                       throw new IllegalStateException(s"Unexpected error ${other.exceptionName} while sending txn marker for $transactionalId")
                   }
                 }
-              }
+              })
             }
 
             if (!abortSending) {