Fix sending lag caused by thread safety

streamnative · Dec 28, 2023 · d52bf75 · d52bf75
1 parent aa2f171
commit d52bf75
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/ReplicaManager.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/ReplicaManager.java
@@ -32,6 +32,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -108,7 +110,7 @@ private static class PendingProduceCallback implements Runnable {
         final CompletableFuture<Map<TopicPartition, ProduceResponse.PartitionResponse>> completableFuture;
         Map<TopicPartition, MemoryRecords> entriesPerPartition;
         @Override
-        public void run() {
+        public synchronized void run() {
             topicPartitionNum.set(0);
             if (completableFuture.isDone()) {
                 // It may be triggered again in DelayedProduceAndFetch