[pinpoint-apm#8865] Correct inconsistent synchronization in volatile …

…field
emeroad · May 17, 2022 · df2b2ee · df2b2ee
1 parent 131da94
commit df2b2ee
Showing 1 changed file with 6 additions and 6 deletions.
diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/service/SendDataToFlinkService.java b/collector/src/main/java/com/navercorp/pinpoint/collector/service/SendDataToFlinkService.java
@@ -20,7 +20,6 @@
 import org.apache.thrift.TBase;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.LogManager;
-import org.springframework.stereotype.Component;
 
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -32,7 +31,7 @@
 public class SendDataToFlinkService {
     private final Logger logger = LogManager.getLogger(this.getClass());
 
-    private volatile List<TcpDataSender<TBase<?, ?>>> flinkTcpDataSenderList = new CopyOnWriteArrayList<>();
+    private volatile List<TcpDataSender<TBase<?, ?>>> dataSenderList = new CopyOnWriteArrayList<>();
     private final AtomicInteger callCount = new AtomicInteger(1);
 
     protected void sendData(TBase<?, ?> data) {
@@ -53,20 +52,21 @@ protected void sendData(TBase<?, ?> data) {
     }
 
     private TcpDataSender<TBase<?, ?>> roundRobinTcpDataSender() {
-        if (flinkTcpDataSenderList.isEmpty()) {
+        final List<TcpDataSender<TBase<?, ?>>> copyList = this.dataSenderList;
+        if (copyList.isEmpty()) {
             return null;
         }
 
         int count = callCount.getAndIncrement();
-        int tcpDataSenderIndex = count % flinkTcpDataSenderList.size();
+        int tcpDataSenderIndex = count % copyList.size();
 
         if (tcpDataSenderIndex < 0) {
             tcpDataSenderIndex = tcpDataSenderIndex * -1;
             callCount.set(0);
         }
 
         try {
-            return flinkTcpDataSenderList.get(tcpDataSenderIndex);
+            return copyList.get(tcpDataSenderIndex);
         } catch (Exception e) {
             logger.warn("not get FlinkTcpDataSender", e);
         }
@@ -75,6 +75,6 @@ protected void sendData(TBase<?, ?> data) {
     }
 
     public void replaceFlinkTcpDataSenderList(List<FlinkTcpDataSender> flinkTcpDataSenderList) {
-        this.flinkTcpDataSenderList = new CopyOnWriteArrayList<>(flinkTcpDataSenderList);
+        this.dataSenderList = new CopyOnWriteArrayList<>(flinkTcpDataSenderList);
     }
 }