From affbaa8bfd3331e358ac5a58e4e286f464baca74 Mon Sep 17 00:00:00 2001
From: jiangyuan <joe469391363@gmail.com>
Date: Sun, 19 Jan 2025 18:44:47 +0800
Subject: [PATCH] Fix ConsumeDriver running status (#748)

---
 CHANGES.md                                            |  1 +
 .../commons/datacarrier/consumer/ConsumeDriver.java   | 11 ++++++++++-
 2 files changed, 11 insertions(+), 1 deletion(-)

diff --git a/CHANGES.md b/CHANGES.md
index 4fd1859d4..2a797ed4c 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -28,6 +28,7 @@ Release Notes.
 * Add Caffeine plugin as optional.
 * Add Undertow 2.1.7.final+ worker thread pool metrics.
 * Support for tracking in spring gateway versions 4.1.2 and above.
+* Fix `ConsumeDriver` running status concurrency issues.
 
 All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/222?closed=1)
 
diff --git a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumeDriver.java b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumeDriver.java
index a6d808fea..853c7fc3b 100644
--- a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumeDriver.java
+++ b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumeDriver.java
@@ -27,7 +27,7 @@ import org.apache.skywalking.apm.commons.datacarrier.buffer.Channels;
  * Pool of consumers <p> Created by wusheng on 2016/10/25.
  */
 public class ConsumeDriver<T> implements IDriver {
-    private boolean running;
+    private volatile boolean running;
     private ConsumerThread[] consumerThreads;
     private Channels<T> channels;
     private ReentrantLock lock;
@@ -88,6 +88,9 @@ public class ConsumeDriver<T> implements IDriver {
         }
         lock.lock();
         try {
+            if (running) {
+                return;
+            }
             this.allocateBuffer2Thread();
             for (ConsumerThread consumerThread : consumerThreads) {
                 consumerThread.start();
@@ -124,8 +127,14 @@ public class ConsumeDriver<T> implements IDriver {
 
     @Override
     public void close(Channels channels) {
+        if (!running) {
+            return;
+        }
         lock.lock();
         try {
+            if (!running) {
+                return;
+            }
             this.running = false;
             for (ConsumerThread consumerThread : consumerThreads) {
                 consumerThread.shutdown();