From 675f0dd03f188b9d1cf67ef77449cd3eef110c5c Mon Sep 17 00:00:00 2001
From: Enrico Olivelli <enrico.olivelli@datastax.com>
Date: Fri, 9 Dec 2022 16:17:26 +0100
Subject: [PATCH] [bugfix][transactions] Prevent
 ConcurrentModificationException in getProducer()

---
 .../kop/coordinator/transaction/TransactionStateManager.java  | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionStateManager.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionStateManager.java
index 65bd3b836a..a77cedd568 100644
--- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionStateManager.java
+++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionStateManager.java
@@ -74,8 +74,8 @@ public class TransactionStateManager {
     @VisibleForTesting
     protected final Set<Integer> leavingPartitions = Sets.newHashSet();
 
-    private final Map<Integer, CompletableFuture<Producer<ByteBuffer>>> txnLogProducerMap = Maps.newHashMap();
-    private final Map<Integer, CompletableFuture<Reader<ByteBuffer>>> txnLogReaderMap = Maps.newHashMap();
+    private final Map<Integer, CompletableFuture<Producer<ByteBuffer>>> txnLogProducerMap = Maps.newConcurrentMap();
+    private final Map<Integer, CompletableFuture<Reader<ByteBuffer>>> txnLogReaderMap = Maps.newConcurrentMap();
 
     // Transaction metadata cache indexed by assigned transaction topic partition ids
     // Map <partitionId, <transactionId, TransactionMetadata>>