MINOR: update TransactionLog#readTxnRecordValue to initialize TransactionMetadata with non-empty topic partitions (#20370)

This is followup PR for #19699.

* Update TransactionLog#readTxnRecordValue to initialize
TransactionMetadata with non-empty topic partitions
* Update `TxnTransitMetadata` comment, because it's not immutable.

Reviewers: TengYao Chi <[email protected]>, Justine Olshan
 <[email protected]>, Kuan-Po Tseng <[email protected]>, Chia-Ping
 Tsai <[email protected]>

Author: FrankYang0529

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

@@ -115,28 +115,25 @@ object TransactionLog {
       val version = buffer.getShort
       if (version >= TransactionLogValue.LOWEST_SUPPORTED_VERSION && version <= TransactionLogValue.HIGHEST_SUPPORTED_VERSION) {
         val value = new TransactionLogValue(new ByteBufferAccessor(buffer), version)
-        val transactionMetadata = new TransactionMetadata(
+        val state = TransactionState.fromId(value.transactionStatus)
+        val tps: util.Set[TopicPartition] = new util.HashSet[TopicPartition]()
+        if (!state.equals(TransactionState.EMPTY))
+          value.transactionPartitions.forEach(partitionsSchema => {
+            partitionsSchema.partitionIds.forEach(partitionId => tps.add(new TopicPartition(partitionsSchema.topic, partitionId.intValue())))
+          })
+        Some(new TransactionMetadata(
           transactionalId,
           value.producerId,
           value.previousProducerId,
           value.nextProducerId,
           value.producerEpoch,
           RecordBatch.NO_PRODUCER_EPOCH,
           value.transactionTimeoutMs,
-          TransactionState.fromId(value.transactionStatus),
-          util.Set.of(),
+          state,
+          tps,
           value.transactionStartTimestampMs,
           value.transactionLastUpdateTimestampMs,
-          TransactionVersion.fromFeatureLevel(value.clientTransactionVersion))
-
-        if (!transactionMetadata.state.equals(TransactionState.EMPTY))
-          value.transactionPartitions.forEach(partitionsSchema => {
-            transactionMetadata.addPartitions(partitionsSchema.partitionIds
-              .stream
-              .map(partitionId => new TopicPartition(partitionsSchema.topic, partitionId.intValue()))
-              .toList)
-          })
-        Some(transactionMetadata)
+          TransactionVersion.fromFeatureLevel(value.clientTransactionVersion)))
       } else throw new IllegalStateException(s"Unknown version $version from the transaction log message value")
     }
   }

transaction-coordinator/src/main/java/org/apache/kafka/coordinator/transaction/TransactionMetadata.java

@@ -117,6 +117,7 @@ public <T> T inLock(Supplier<T> function) {
         }
     }
 
+    // VisibleForTesting
     public void addPartitions(Collection<TopicPartition> partitions) {
         topicPartitions.addAll(partitions);
     }
@@ -500,13 +501,15 @@ public String transactionalId() {
         return transactionalId;
     }
 
+    // VisibleForTesting
     public void setProducerId(long producerId) {
         this.producerId = producerId;
     }
     public long producerId() {
         return producerId;
     }
 
+    // VisibleForTesting
     public void setPrevProducerId(long prevProducerId) {
         this.prevProducerId = prevProducerId;
     }
@@ -534,6 +537,7 @@ public int txnTimeoutMs() {
         return txnTimeoutMs;
     }
 
+    // VisibleForTesting
     public void state(TransactionState state) {
         this.state = state;
     }
@@ -550,6 +554,7 @@ public long txnStartTimestamp() {
         return txnStartTimestamp;
     }
 
+    // VisibleForTesting
     public void txnLastUpdateTimestamp(long txnLastUpdateTimestamp) {
         this.txnLastUpdateTimestamp = txnLastUpdateTimestamp;
     }

transaction-coordinator/src/main/java/org/apache/kafka/coordinator/transaction/TxnTransitMetadata.java

@@ -22,7 +22,7 @@
 import java.util.HashSet;
 
 /**
- * Immutable object representing the target transition of the transaction metadata
+ * Represent the target transition of the transaction metadata. The topicPartitions field is mutable.
  */
 public record TxnTransitMetadata(
         long producerId,