[pulsar] 02/02: [fix][broker] Copy command fields and fix potential thread-safety in ServerCnx (#19517)

[nicoloboschi]

This is an automated email from the ASF dual-hosted git repository.

nicoloboschi pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git

commit 682d045b76a5a55ba368e6978b9692ad784c9d4c
Author: Nicolò Boschi <boschi1...@gmail.com>
AuthorDate: Thu Feb 23 13:54:27 2023 +0100

    [fix][broker] Copy command fields and fix potential thread-safety in 
ServerCnx (#19517)
    
    (cherry picked from commit 0bb0f6b786d115a7405867b701521cd4a49340c5)
---
 .../src/main/java/org/apache/pulsar/broker/service/ServerCnx.java  | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index b25d9514c1c..f3f1bc69636 100644
--- 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -2326,10 +2326,11 @@ public class ServerCnx extends PulsarHandler implements 
TransportCnx {
         final TxnID txnID = new TxnID(command.getTxnidMostBits(), 
command.getTxnidLeastBits());
         final TransactionCoordinatorID tcId = 
TransactionCoordinatorID.get(command.getTxnidMostBits());
         final long requestId = command.getRequestId();
+        final List<String> partitionsList = command.getPartitionsList();
         if (log.isDebugEnabled()) {
-            command.getPartitionsList().forEach(partion ->
+            partitionsList.forEach(partition ->
                     log.debug("Receive add published partition to txn request 
{} "
-                            + "from {} with txnId {}, topic: [{}]", requestId, 
remoteAddress, txnID, partion));
+                            + "from {} with txnId {}, topic: [{}]", requestId, 
remoteAddress, txnID, partition));
         }
 
         if (!checkTransactionEnableAndSendError(requestId)) {
@@ -2344,7 +2345,7 @@ public class ServerCnx extends PulsarHandler implements 
TransportCnx {
                         return failedFutureTxnNotOwned(txnID);
                     }
                     return transactionMetadataStoreService
-                            .addProducedPartitionToTxn(txnID, 
command.getPartitionsList());
+                            .addProducedPartitionToTxn(txnID, partitionsList);
                 })
                 .whenComplete((v, ex) -> {
                     if (ex == null) {