Repository: alibaba/Sentinel (java)

Issue:
Connection leak risk when using Nacos data-source | Nacos 数据源 close 存在连接泄漏问题
## Issue Description

Type: *bug report*

### Describe what happened (or what feature you want)

when `com.alibaba.csp.sentinel.datasource.nacos.NacosDataSource ` init, it will create a new `ConfigService` and cached it.
But when NacosDataSource close, the `ConfigService` is not be shutdown so that the connections can threads may be not closed.

Once `com.alibaba.csp.sentinel.datasource.nacos.NacosDataSource ` rebuild, the connection and thread will leak.

### Describe what you expected to happen

shutdown `ConfigService` when `com.alibaba.csp.sentinel.datasource.nacos.NacosDataSource ` close.

### Anything else we need to know?

The other and recommended fixed way is re-use the `ConfigService`  for all datasources. Now sentinel will create one connection to nacos for each rule. 
In fact, multiple rule can be listened by one `ConfigService`.



Pull request:
Fix issue: connection leak risk when using Nacos data-source.
### Describe what this PR does / why we need it
Connection in `NacosDataSource` exists leak risk.

### Does this pull request fix one issue?
Fixes #2953

### Describe how you did it
`ConfigService#shutDown()` when `NacosDataSource#close()`.

### Describe how to verify it


### Special notes for reviews


Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 59fead0f..e576e7cb 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -14,12 +14,13 @@ jobs:
         java: [8, 11]
     steps:
       - name: Checkout
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
 
       - name: Setup Java
-        uses: actions/setup-java@v1
+        uses: actions/setup-java@v3
         with:
           java-version: ${{ matrix.java }}
+          distribution: 'adopt'
           architecture: x64
 
       - name: Test with Maven
@@ -29,4 +30,4 @@ jobs:
         run: mvn install -DskipTests=true -Dmaven.javadoc.skip=true -B -V -DminimumPriority=1
 
       - name: Run Codecov
-        run: bash <(curl -s https://codecov.io/bash)
\ No newline at end of file
+        run: bash <(curl -s https://codecov.io/bash)
diff --git a/sentinel-adapter/sentinel-grpc-adapter/pom.xml b/sentinel-adapter/sentinel-grpc-adapter/pom.xml
index 04566d53..5ffe958e 100755
--- a/sentinel-adapter/sentinel-grpc-adapter/pom.xml
+++ b/sentinel-adapter/sentinel-grpc-adapter/pom.xml
@@ -12,7 +12,8 @@
     <packaging>jar</packaging>
 
     <properties>
-        <grpc.version>1.30.2</grpc.version>
+        <protobuf.version>3.21.9</protobuf.version>
+        <grpc.version>1.51.0</grpc.version>
     </properties>
 
     <dependencies>
@@ -79,7 +80,7 @@
                 <artifactId>protobuf-maven-plugin</artifactId>
                 <version>0.5.1</version>
                 <configuration>
-                    <protocArtifact>com.google.protobuf:protoc:3.5.1-1:exe:${os.detected.classifier}</protocArtifact>
+                    <protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}</protocArtifact>
                     <pluginId>grpc-java</pluginId>
                     <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}
                     </pluginArtifact>
diff --git a/sentinel-cluster/sentinel-cluster-server-envoy-rls/pom.xml b/sentinel-cluster/sentinel-cluster-server-envoy-rls/pom.xml
index 09231216..30bfd70d 100644
--- a/sentinel-cluster/sentinel-cluster-server-envoy-rls/pom.xml
+++ b/sentinel-cluster/sentinel-cluster-server-envoy-rls/pom.xml
@@ -15,8 +15,8 @@
         <java.source.version>1.8</java.source.version>
         <java.target.version>1.8</java.target.version>
 
-        <protobuf.version>3.10.0</protobuf.version>
-        <grpc.version>1.30.2</grpc.version>
+        <protobuf.version>3.21.9</protobuf.version>
+        <grpc.version>1.51.0</grpc.version>
 
         <maven.shade.version>3.2.1</maven.shade.version>
     </properties>
diff --git a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/FlowRuleUtil.java b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/FlowRuleUtil.java
index f349e952..b607ef6b 100644
--- a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/FlowRuleUtil.java
+++ b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/FlowRuleUtil.java
@@ -19,7 +19,7 @@ import com.alibaba.csp.sentinel.log.RecordLog;
 import com.alibaba.csp.sentinel.slots.block.ClusterRuleConstant;
 import com.alibaba.csp.sentinel.slots.block.RuleConstant;
 import com.alibaba.csp.sentinel.slots.block.flow.controller.DefaultController;
-import com.alibaba.csp.sentinel.slots.block.flow.controller.RateLimiterController;
+import com.alibaba.csp.sentinel.slots.block.flow.controller.ThrottlingController;
 import com.alibaba.csp.sentinel.slots.block.flow.controller.WarmUpController;
 import com.alibaba.csp.sentinel.slots.block.flow.controller.WarmUpRateLimiterController;
 import com.alibaba.csp.sentinel.util.StringUtil;
@@ -136,7 +136,7 @@ public final class FlowRuleUtil {
                     return new WarmUpController(rule.getCount(), rule.getWarmUpPeriodSec(),
                             ColdFactorProperty.coldFactor);
                 case RuleConstant.CONTROL_BEHAVIOR_RATE_LIMITER:
-                    return new RateLimiterController(rule.getMaxQueueingTimeMs(), rule.getCount());
+                    return new ThrottlingController(rule.getMaxQueueingTimeMs(), rule.getCount());
                 case RuleConstant.CONTROL_BEHAVIOR_WARM_UP_RATE_LIMITER:
                     return new WarmUpRateLimiterController(rule.getCount(), rule.getWarmUpPeriodSec(),
                             rule.getMaxQueueingTimeMs(), ColdFactorProperty.coldFactor);
diff --git a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/controller/RateLimiterController.java b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/controller/RateLimiterController.java
deleted file mode 100755
index ca07b028..00000000
--- a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/controller/RateLimiterController.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * Copyright 1999-2018 Alibaba Group Holding Ltd.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package com.alibaba.csp.sentinel.slots.block.flow.controller;
-
-import java.util.concurrent.atomic.AtomicLong;
-
-import com.alibaba.csp.sentinel.slots.block.flow.TrafficShapingController;
-
-import com.alibaba.csp.sentinel.util.TimeUtil;
-import com.alibaba.csp.sentinel.node.Node;
-
-/**
- * @author jialiang.linjl
- */
-public class RateLimiterController implements TrafficShapingController {
-
-    private final int maxQueueingTimeMs;
-    private final double count;
-
-    private final AtomicLong latestPassedTime = new AtomicLong(-1);
-
-    public RateLimiterController(int timeOut, double count) {
-        this.maxQueueingTimeMs = timeOut;
-        this.count = count;
-    }
-
-    @Override
-    public boolean canPass(Node node, int acquireCount) {
-        return canPass(node, acquireCount, false);
-    }
-
-    @Override
-    public boolean canPass(Node node, int acquireCount, boolean prioritized) {
-        // Pass when acquire count is less or equal than 0.
-        if (acquireCount <= 0) {
-            return true;
-        }
-        // Reject when count is less or equal than 0.
-        // Otherwise,the costTime will be max of long and waitTime will overflow in some cases.
-        if (count <= 0) {
-            return false;
-        }
-
-        long currentTime = TimeUtil.currentTimeMillis();
-        // Calculate the interval between every two requests.
-        long costTime = Math.round(1.0 * (acquireCount) / count * 1000);
-
-        // Expected pass time of this request.
-        long expectedTime = costTime + latestPassedTime.get();
-
-        if (expectedTime <= currentTime) {
-            // Contention may exist here, but it's okay.
-            latestPassedTime.set(currentTime);
-            return true;
-        } else {
-            // Calculate the time to wait.
-            long waitTime = costTime + latestPassedTime.get() - TimeUtil.currentTimeMillis();
-            if (waitTime > maxQueueingTimeMs) {
-                return false;
-            } else {
-                long oldTime = latestPassedTime.addAndGet(costTime);
-                try {
-                    waitTime = oldTime - TimeUtil.currentTimeMillis();
-                    if (waitTime > maxQueueingTimeMs) {
-                        latestPassedTime.addAndGet(-costTime);
-                        return false;
-                    }
-                    // in race condition waitTime may <= 0
-                    if (waitTime > 0) {
-                        Thread.sleep(waitTime);
-                    }
-                    return true;
-                } catch (InterruptedException e) {
-                }
-            }
-        }
-        return false;
-    }
-
-}
diff --git a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/controller/ThrottlingController.java b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/controller/ThrottlingController.java
new file mode 100644
index 00000000..64eaefc5
--- /dev/null
+++ b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/controller/ThrottlingController.java
@@ -0,0 +1,167 @@
+/*
+ * Copyright 1999-2022 Alibaba Group Holding Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alibaba.csp.sentinel.slots.block.flow.controller;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.LockSupport;
+
+import com.alibaba.csp.sentinel.node.Node;
+import com.alibaba.csp.sentinel.slots.block.flow.TrafficShapingController;
+import com.alibaba.csp.sentinel.util.AssertUtil;
+import com.alibaba.csp.sentinel.util.TimeUtil;
+
+/**
+ * @author Eric Zhao
+ * @author jialiang.linjl
+ * @since 2.0
+ */
+public class ThrottlingController implements TrafficShapingController {
+
+    // Refactored from legacy RateLimitController of Sentinel 1.x.
+
+    private static final long MS_TO_NS_OFFSET = TimeUnit.MILLISECONDS.toNanos(1);
+
+    private final int maxQueueingTimeMs;
+    private final int statDurationMs;
+
+    private final double count;
+    private final boolean useNanoSeconds;
+
+    private final AtomicLong latestPassedTime = new AtomicLong(-1);
+
+    public ThrottlingController(int queueingTimeoutMs, double maxCountPerStat) {
+        this(queueingTimeoutMs, maxCountPerStat, 1000);
+    }
+
+    public ThrottlingController(int queueingTimeoutMs, double maxCountPerStat, int statDurationMs) {
+        AssertUtil.assertTrue(statDurationMs > 0, "statDurationMs should be positive");
+        AssertUtil.assertTrue(maxCountPerStat >= 0, "maxCountPerStat should be >= 0");
+        AssertUtil.assertTrue(queueingTimeoutMs >= 0, "queueingTimeoutMs should be >= 0");
+        this.maxQueueingTimeMs = queueingTimeoutMs;
+        this.count = maxCountPerStat;
+        this.statDurationMs = statDurationMs;
+        // Use nanoSeconds when durationMs%count != 0 or count/durationMs> 1 (to be accurate)
+        if (maxCountPerStat > 0) {
+            this.useNanoSeconds = statDurationMs % Math.round(maxCountPerStat) != 0 || maxCountPerStat / statDurationMs > 1;
+        } else {
+            this.useNanoSeconds = false;
+        }
+    }
+
+    @Override
+    public boolean canPass(Node node, int acquireCount) {
+        return canPass(node, acquireCount, false);
+    }
+
+    private boolean checkPassUsingNanoSeconds(int acquireCount, double maxCountPerStat) {
+        final long maxQueueingTimeNs = maxQueueingTimeMs * MS_TO_NS_OFFSET;
+        long currentTime = System.nanoTime();
+        // Calculate the interval between every two requests.
+        final long costTimeNs = Math.round(1.0d * MS_TO_NS_OFFSET * statDurationMs * acquireCount / maxCountPerStat);
+
+        // Expected pass time of this request.
+        long expectedTime = costTimeNs + latestPassedTime.get();
+
+        if (expectedTime <= currentTime) {
+            // Contention may exist here, but it's okay.
+            latestPassedTime.set(currentTime);
+            return true;
+        } else {
+            final long curNanos = System.nanoTime();
+            // Calculate the time to wait.
+            long waitTime = costTimeNs + latestPassedTime.get() - curNanos;
+            if (waitTime > maxQueueingTimeNs) {
+                return false;
+            }
+
+            long oldTime = latestPassedTime.addAndGet(costTimeNs);
+            waitTime = oldTime - curNanos;
+            if (waitTime > maxQueueingTimeNs) {
+                latestPassedTime.addAndGet(-costTimeNs);
+                return false;
+            }
+            // in race condition waitTime may <= 0
+            if (waitTime > 0) {
+                sleepNanos(waitTime);
+            }
+            return true;
+        }
+    }
+
+    private boolean checkPassUsingCachedMs(int acquireCount, double maxCountPerStat) {
+        long currentTime = TimeUtil.currentTimeMillis();
+        // Calculate the interval between every two requests.
+        long costTime = Math.round(1.0d * statDurationMs * acquireCount / maxCountPerStat);
+
+        // Expected pass time of this request.
+        long expectedTime = costTime + latestPassedTime.get();
+
+        if (expectedTime <= currentTime) {
+            // Contention may exist here, but it's okay.
+            latestPassedTime.set(currentTime);
+            return true;
+        } else {
+            // Calculate the time to wait.
+            long waitTime = costTime + latestPassedTime.get() - TimeUtil.currentTimeMillis();
+            if (waitTime > maxQueueingTimeMs) {
+                return false;
+            }
+
+            long oldTime = latestPassedTime.addAndGet(costTime);
+            waitTime = oldTime - TimeUtil.currentTimeMillis();
+            if (waitTime > maxQueueingTimeMs) {
+                latestPassedTime.addAndGet(-costTime);
+                return false;
+            }
+            // in race condition waitTime may <= 0
+            if (waitTime > 0) {
+                sleepMs(waitTime);
+            }
+            return true;
+        }
+    }
+
+    @Override
+    public boolean canPass(Node node, int acquireCount, boolean prioritized) {
+        // Pass when acquire count is less or equal than 0.
+        if (acquireCount <= 0) {
+            return true;
+        }
+        // Reject when count is less or equal than 0.
+        // Otherwise, the costTime will be max of long and waitTime will overflow in some cases.
+        if (count <= 0) {
+            return false;
+        }
+        if (useNanoSeconds) {
+            return checkPassUsingNanoSeconds(acquireCount, this.count);
+        } else {
+            return checkPassUsingCachedMs(acquireCount, this.count);
+        }
+    }
+
+    private void sleepMs(long ms) {
+        try {
+            Thread.sleep(ms);
+        } catch (InterruptedException e) {
+        }
+    }
+
+    private void sleepNanos(long ns) {
+        LockSupport.parkNanos(ns);
+    }
+
+}
diff --git a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/logger/LogSlot.java b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/logger/LogSlot.java
index e70457f5..f521e84e 100755
--- a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/logger/LogSlot.java
+++ b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/logger/LogSlot.java
@@ -38,12 +38,11 @@ public class LogSlot extends AbstractLinkedProcessorSlot<DefaultNode> {
             fireEntry(context, resourceWrapper, obj, count, prioritized, args);
         } catch (BlockException e) {
             EagleEyeLogUtil.log(resourceWrapper.getName(), e.getClass().getSimpleName(), e.getRuleLimitApp(),
-                context.getOrigin(), e.getRule().getId(), count);
+                context.getOrigin(), e.getRule() != null ? e.getRule().getId() : null, count);
             throw e;
         } catch (Throwable e) {
             RecordLog.warn("Unexpected entry exception", e);
         }
-
     }
 
     @Override
diff --git a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/util/AssertUtil.java b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/util/AssertUtil.java
index 545bcc4d..78935af9 100644
--- a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/util/AssertUtil.java
+++ b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/util/AssertUtil.java
@@ -24,7 +24,19 @@ import java.util.Collection;
  */
 public class AssertUtil {
 
-    private AssertUtil(){}
+    private AssertUtil() {}
+
+    public static void assertNotNull(Object object, String message) {
+        if (object == null) {
+            throw new IllegalArgumentException(message);
+        }
+    }
+
+    public static void assertTrue(boolean value, String message) {
+        if (!value) {
+            throw new IllegalArgumentException(message);
+        }
+    }
 
     public static void notEmpty(String string, String message) {
         if (StringUtil.isEmpty(string)) {
diff --git a/sentinel-dashboard/Dockerfile b/sentinel-dashboard/Dockerfile
new file mode 100644
index 00000000..3a3412db
--- /dev/null
+++ b/sentinel-dashboard/Dockerfile
@@ -0,0 +1,20 @@
+FROM amd64/buildpack-deps:buster-curl as installer
+
+ARG SENTINEL_VERSION=1.8.6
+
+RUN set -x \
+    && curl -SL --output /home/sentinel-dashboard.jar https://github.com/alibaba/Sentinel/releases/download/${SENTINEL_VERSION}/sentinel-dashboard-${SENTINEL_VERSION}.jar
+
+FROM openjdk:8-jre-slim
+
+# copy sentinel jar
+COPY --from=installer ["/home/sentinel-dashboard.jar", "/home/sentinel-dashboard.jar"]
+
+ENV JAVA_OPTS '-Dserver.port=8080 -Dcsp.sentinel.dashboard.server=localhost:8080'
+
+RUN chmod -R +x /home/sentinel-dashboard.jar
+
+EXPOSE 8080
+
+CMD java ${JAVA_OPTS} -jar /home/sentinel-dashboard.jar
+
diff --git a/sentinel-dashboard/README.md b/sentinel-dashboard/README.md
index ad8b91f9..e3597725 100755
--- a/sentinel-dashboard/README.md
+++ b/sentinel-dashboard/README.md
@@ -58,4 +58,10 @@ java -Dserver.port=8080 \
 客户端正确配置并启动后，会**在初次调用后**主动向控制台发送心跳包，汇报自己的存在；
 控制台收到客户端心跳包之后，会在左侧导航栏中显示该客户端信息。如果控制台能够看到客户端的机器信息，则表明客户端接入成功了。
 
+## 6. 构建Docker镜像
+
+```
+docker build --build-arg SENTINEL_VERSION=1.8.6 -t ${REGISTRY}/sentinel-dashboard:v1.8.6 .
+```
+
 更多：[控制台功能介绍](./Sentinel_Dashboard_Feature.md)。
\ No newline at end of file
diff --git a/sentinel-extension/sentinel-datasource-etcd/src/main/java/com/alibaba/csp/sentinel/datasource/etcd/EtcdDataSource.java b/sentinel-extension/sentinel-datasource-etcd/src/main/java/com/alibaba/csp/sentinel/datasource/etcd/EtcdDataSource.java
index bc72e823..33684eb8 100644
--- a/sentinel-extension/sentinel-datasource-etcd/src/main/java/com/alibaba/csp/sentinel/datasource/etcd/EtcdDataSource.java
+++ b/sentinel-extension/sentinel-datasource-etcd/src/main/java/com/alibaba/csp/sentinel/datasource/etcd/EtcdDataSource.java
@@ -88,7 +88,8 @@ public class EtcdDataSource<T> extends AbstractDataSource<String, T> {
                 WatchEvent.EventType eventType = watchEvent.getEventType();
                 if (eventType == WatchEvent.EventType.PUT) {
                     try {
-                        T newValue = loadConfig();
+                        String newValueJson = watchEvent.getKeyValue().getValue().toString(charset);
+                        T newValue = parser.convert(newValueJson);
                         getProperty().updateValue(newValue);
                     } catch (Exception e) {
                         RecordLog.warn("[EtcdDataSource] Failed to update config", e);
diff --git a/sentinel-extension/sentinel-datasource-nacos/src/main/java/com/alibaba/csp/sentinel/datasource/nacos/NacosDataSource.java b/sentinel-extension/sentinel-datasource-nacos/src/main/java/com/alibaba/csp/sentinel/datasource/nacos/NacosDataSource.java
index ee5c7190..caade5cf 100644
--- a/sentinel-extension/sentinel-datasource-nacos/src/main/java/com/alibaba/csp/sentinel/datasource/nacos/NacosDataSource.java
+++ b/sentinel-extension/sentinel-datasource-nacos/src/main/java/com/alibaba/csp/sentinel/datasource/nacos/NacosDataSource.java
@@ -145,6 +145,12 @@ public class NacosDataSource<T> extends AbstractDataSource<String, T> {
     public void close() {
         if (configService != null) {
             configService.removeListener(dataId, groupId, configListener);
+            try {
+                configService.shutDown();
+            } catch (Exception e) {
+                RecordLog.warn("[NacosDataSource] Error occurred when closing Nacos data source", e);
+                e.printStackTrace();
+            }
         }
         pool.shutdownNow();
     }
diff --git a/sentinel-extension/sentinel-datasource-opensergo/src/main/java/com/alibaba/csp/sentinel/datasource/OpenSergoDataSourceGroup.java b/sentinel-extension/sentinel-datasource-opensergo/src/main/java/com/alibaba/csp/sentinel/datasource/OpenSergoDataSourceGroup.java
index c07c6a5e..83944bfc 100644
--- a/sentinel-extension/sentinel-datasource-opensergo/src/main/java/com/alibaba/csp/sentinel/datasource/OpenSergoDataSourceGroup.java
+++ b/sentinel-extension/sentinel-datasource-opensergo/src/main/java/com/alibaba/csp/sentinel/datasource/OpenSergoDataSourceGroup.java
@@ -31,6 +31,7 @@ import com.alibaba.csp.sentinel.util.AssertUtil;
 import io.opensergo.ConfigKind;
 import io.opensergo.OpenSergoClient;
 import io.opensergo.proto.fault_tolerance.v1.CircuitBreakerStrategy;
+import io.opensergo.proto.fault_tolerance.v1.ConcurrencyLimitStrategy;
 import io.opensergo.proto.fault_tolerance.v1.FaultToleranceRule;
 import io.opensergo.proto.fault_tolerance.v1.RateLimitStrategy;
 import io.opensergo.proto.fault_tolerance.v1.ThrottlingStrategy;
@@ -270,8 +271,7 @@ public class OpenSergoDataSourceGroup {
                 case THROTTLING_STRATEGY:
                     return ruleAggregator.updateThrottlingStrategy((List<ThrottlingStrategy>) data);
                 case CONCURRENCY_LIMIT_STRATEGY:
-                    // TODO
-                    return false;
+                    return ruleAggregator.updateConcurrencyLimitStrategy((List<ConcurrencyLimitStrategy>) data);
                 default:
                     // Type mismatch
                     return false;
diff --git a/sentinel-extension/sentinel-datasource-opensergo/src/main/java/com/alibaba/csp/sentinel/datasource/OpenSergoRuleAggregator.java b/sentinel-extension/sentinel-datasource-opensergo/src/main/java/com/alibaba/csp/sentinel/datasource/OpenSergoRuleAggregator.java
index cc0390ee..2839f5cd 100644
--- a/sentinel-extension/sentinel-datasource-opensergo/src/main/java/com/alibaba/csp/sentinel/datasource/OpenSergoRuleAggregator.java
+++ b/sentinel-extension/sentinel-datasource-opensergo/src/main/java/com/alibaba/csp/sentinel/datasource/OpenSergoRuleAggregator.java
@@ -114,6 +114,16 @@ public class OpenSergoRuleAggregator {
         return handleCircuitBreakerRuleUpdate();
     }
 
+    public synchronized boolean updateConcurrencyLimitStrategy(List<ConcurrencyLimitStrategy> strategies) {
+        Map<String, ConcurrencyLimitStrategy> map = new HashMap<>(4);
+        if (strategies != null && !strategies.isEmpty()) {
+            strategies.forEach(s -> map.put(s.getName(), s));
+        }
+        this.concurrencyLimitStrategyMap = map;
+
+        return handleFlowRuleUpdate();
+    }
+
     private boolean handleFlowRuleUpdate() {
         List<FlowRule> rules = new ArrayList<>();
 
@@ -121,9 +131,12 @@ public class OpenSergoRuleAggregator {
             ftRuleMapByStrategyKind.get(ConfigKind.RATE_LIMIT_STRATEGY.getSimpleKindName()), rateLimitStrategyMap);
         List<FlowRule> rulesFromThrottlingStrategies = assembleFlowRulesFromThrottlingStrategies(
             ftRuleMapByStrategyKind.get(ConfigKind.THROTTLING_STRATEGY.getSimpleKindName()), throttlingStrategyMap);
+        List<FlowRule> rulesConcurrencyLimitStrategies = assembleFlowRulesFromConcurrencyLimitStrategies(
+            ftRuleMapByStrategyKind.get(ConfigKind.CONCURRENCY_LIMIT_STRATEGY.getSimpleKindName()), concurrencyLimitStrategyMap);
 
         rules.addAll(rulesFromRateLimitStrategies);
         rules.addAll(rulesFromThrottlingStrategies);
+        rules.addAll(rulesConcurrencyLimitStrategies);
 
         // Update rules to upstream data-source.
         return dataSourceMap.get(OpenSergoSentinelConstants.KIND_FLOW_RULE).updateValue(rules);
@@ -174,6 +187,42 @@ public class OpenSergoRuleAggregator {
         return rules;
     }
 
+    private List<FlowRule> assembleFlowRulesFromConcurrencyLimitStrategies(List<FaultToleranceRule> ftRules,
+                                                                    Map<String, ConcurrencyLimitStrategy> concurrencyLimitStrategyMap) {
+        List<FlowRule> rules = new ArrayList<>();
+        if (ftRules == null || ftRules.isEmpty()) {
+            return rules;
+        }
+        for (FaultToleranceRule ftRule : ftRules) {
+            List<ConcurrencyLimitStrategy> strategies = ftRule.getStrategiesList().stream()
+                .filter(e -> e.getKind().equals(ConfigKind.CONCURRENCY_LIMIT_STRATEGY.getSimpleKindName()))
+                .map(e -> concurrencyLimitStrategyMap.get(e.getName()))
+                .filter(Objects::nonNull)
+                .collect(Collectors.toList());
+            if (strategies.isEmpty()) {
+                continue;
+            }
+
+            for (FaultToleranceRuleTargetRef targetRef : ftRule.getTargetsList()) {
+                String resourceName = targetRef.getTargetResourceName();
+
+                for (ConcurrencyLimitStrategy strategy : strategies) {
+                    FlowRule flowRule = new FlowRule(resourceName);
+                    try {
+                        flowRule = fillFlowRuleWithConcurrencyLimitStrategy(flowRule, strategy);
+                        if (flowRule != null) {
+                            rules.add(flowRule);
+                        }
+                    } catch (Exception ex) {
+                        RecordLog.warn("Ignoring OpenSergo ConcurrencyLimitStrategy due to covert failure, "
+                            + "resourceName={}, strategy={}", resourceName, strategy);
+                    }
+                }
+            }
+        }
+        return rules;
+    }
+
     private List<FlowRule> assembleFlowRulesFromThrottlingStrategies(List<FaultToleranceRule> ftRules,
                                                                      Map<String, ThrottlingStrategy> throttlingStrategyMap) {
         List<FlowRule> rules = new ArrayList<>();
@@ -255,6 +304,19 @@ public class OpenSergoRuleAggregator {
         return rule;
     }
 
+    private FlowRule fillFlowRuleWithConcurrencyLimitStrategy(FlowRule rule, ConcurrencyLimitStrategy strategy) {
+        if (rule == null || strategy == null) {
+            return rule;
+        }
+        rule.setCount(strategy.getMaxConcurrency());
+        rule.setGrade(RuleConstant.FLOW_GRADE_THREAD);
+        rule.setClusterMode(false);
+        rule.setStrategy(RuleConstant.STRATEGY_DIRECT);
+        rule.setControlBehavior(RuleConstant.CONTROL_BEHAVIOR_DEFAULT);
+
+        return rule;
+    }
+
     private FlowRule fillFlowRuleWithThrottlingStrategy(FlowRule rule, ThrottlingStrategy strategy) {
         if (rule == null || strategy == null) {
             return rule;
diff --git a/sentinel-extension/sentinel-metric-exporter/src/main/java/com/alibaba/csp/sentinel/metric/exporter/jmx/MetricBeanWriter.java b/sentinel-extension/sentinel-metric-exporter/src/main/java/com/alibaba/csp/sentinel/metric/exporter/jmx/MetricBeanWriter.java
index 974d0692..68f3f408 100644
--- a/sentinel-extension/sentinel-metric-exporter/src/main/java/com/alibaba/csp/sentinel/metric/exporter/jmx/MetricBeanWriter.java
+++ b/sentinel-extension/sentinel-metric-exporter/src/main/java/com/alibaba/csp/sentinel/metric/exporter/jmx/MetricBeanWriter.java
@@ -20,12 +20,13 @@ package com.alibaba.csp.sentinel.metric.exporter.jmx;
 import com.alibaba.csp.sentinel.config.SentinelConfig;
 import com.alibaba.csp.sentinel.log.RecordLog;
 import com.alibaba.csp.sentinel.node.metric.MetricNode;
+import com.alibaba.csp.sentinel.util.StringUtil;
 
-import java.util.HashSet;
+import javax.management.ObjectName;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Set;
+import java.util.regex.Pattern;
 
 /**
  * the metric bean writer, it provides {@link MetricBeanWriter#write} method for register the
@@ -40,6 +41,8 @@ public class MetricBeanWriter {
     private final MBeanRegistry mBeanRegistry = MBeanRegistry.getInstance();
     
     private static final String DEFAULT_APP_NAME = "sentinel-application";
+
+    private static final Pattern SPECIAL_CHARACTER_PATTERN = Pattern.compile("[*?=:\"\n]");
     
     /**
      * write the MetricNode value to MetricBean
@@ -66,10 +69,14 @@ public class MetricBeanWriter {
             appName = DEFAULT_APP_NAME;
         }
         long version = System.currentTimeMillis();
-        // set or update the new value
+        // set or update the new metric value
         for (MetricNode metricNode : map.values()) {
-            final String mBeanName = "Sentinel:type=" + appName + ",name=\"" + metricNode.getResource()
-                    +"\",classification=\"" + metricNode.getClassification() +"\"";
+            // Fix JMX Metrics export error: https://github.com/alibaba/Sentinel/issues/2989
+            // Without escape, it will throw "cannot add mbean for pattern name" or "invalid character in value part of property" exception
+            String resourceName = escapeSpecialCharacter(metricNode.getResource());
+            final String mBeanName = "Sentinel:type=Metric,resource=" + resourceName
+                    +",classification=" + metricNode.getClassification()
+                    +",appName=" + appName;
             MetricBean metricBean = mBeanRegistry.findMBean(mBeanName);
             if (metricBean != null) {
                 metricBean.setValueFromNode(metricNode);
@@ -94,4 +101,17 @@ public class MetricBeanWriter {
             }
         }
     }
+
+    /**
+     * escape only when arg has special character eg.(*,?,\n,\")
+     *
+     * @param resourceName need escape resource name
+     * @return escaped characters
+     */
+    public static String escapeSpecialCharacter(String resourceName) {
+        if (StringUtil.isBlank(resourceName) || !SPECIAL_CHARACTER_PATTERN.matcher(resourceName).find()) {
+            return resourceName;
+        }
+        return ObjectName.quote(resourceName);
+    }
 }


Test patch:
diff --git a/sentinel-core/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/controller/RateLimiterControllerTest.java b/sentinel-core/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/controller/ThrottlingControllerTest.java
old mode 100755
new mode 100644
similarity index 63%
rename from sentinel-core/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/controller/RateLimiterControllerTest.java
rename to sentinel-core/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/controller/ThrottlingControllerTest.java
index 0b83f5c0..6aa6a7ad
--- a/sentinel-core/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/controller/RateLimiterControllerTest.java
+++ b/sentinel-core/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/controller/ThrottlingControllerTest.java
@@ -1,11 +1,11 @@
 /*
- * Copyright 1999-2018 Alibaba Group Holding Ltd.
+ * Copyright 1999-2022 Alibaba Group Holding Ltd.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *      http://www.apache.org/licenses/LICENSE-2.0
+ *      https://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -15,26 +15,26 @@
  */
 package com.alibaba.csp.sentinel.slots.block.flow.controller;
 
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.mock;
-
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import com.alibaba.csp.sentinel.node.Node;
+import com.alibaba.csp.sentinel.util.TimeUtil;
+
 import org.junit.Test;
 
-import com.alibaba.csp.sentinel.util.TimeUtil;
-import com.alibaba.csp.sentinel.node.Node;
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.mock;
 
 /**
+ * @author Eric Zhao
  * @author jialiang.linjl
  */
-public class RateLimiterControllerTest {
+public class ThrottlingControllerTest {
 
     @Test
-    public void testPaceController_normal() throws InterruptedException {
-        RateLimiterController paceController = new RateLimiterController(500, 10d);
+    public void testThrottlingControllerNormal() throws InterruptedException {
+        ThrottlingController paceController = new ThrottlingController(500, 10d);
         Node node = mock(Node.class);
 
         long start = TimeUtil.currentTimeMillis();
@@ -46,12 +46,12 @@ public class RateLimiterControllerTest {
     }
 
     @Test
-    public void testPaceController_timeout() throws InterruptedException {
-        final RateLimiterController paceController = new RateLimiterController(500, 10d);
+    public void testThrottlingControllerQueueTimeout() throws InterruptedException {
+        final ThrottlingController paceController = new ThrottlingController(500, 10d);
         final Node node = mock(Node.class);
 
-        final AtomicInteger passcount = new AtomicInteger();
-        final AtomicInteger blockcount = new AtomicInteger();
+        final AtomicInteger passCount = new AtomicInteger();
+        final AtomicInteger blockCount = new AtomicInteger();
         final CountDownLatch countDown = new CountDownLatch(1);
 
         final AtomicInteger done = new AtomicInteger();
@@ -62,9 +62,9 @@ public class RateLimiterControllerTest {
                     boolean pass = paceController.canPass(node, 1);
 
                     if (pass) {
-                        passcount.incrementAndGet();
+                        passCount.incrementAndGet();
                     } else {
-                        blockcount.incrementAndGet();
+                        blockCount.incrementAndGet();
                     }
                     done.incrementAndGet();
 
@@ -73,21 +73,20 @@ public class RateLimiterControllerTest {
                     }
                 }
 
-            }, "Thread " + i);
+            }, "Thread-TestThrottlingControllerQueueTimeout-" + i);
             thread.start();
         }
-
         countDown.await();
-        System.out.println("pass:" + passcount.get());
-        System.out.println("block" + blockcount.get());
-        System.out.println("done" + done.get());
-        assertTrue(blockcount.get() > 0);
 
+        System.out.println("pass: " + passCount.get());
+        System.out.println("block: " + blockCount.get());
+        System.out.println("done: " + done.get());
+        assertTrue(blockCount.get() > 0);
     }
 
     @Test
-    public void testPaceController_zeroattack() throws InterruptedException {
-        RateLimiterController paceController = new RateLimiterController(500, 0d);
+    public void testThrottlingControllerZeroThreshold() throws InterruptedException {
+        ThrottlingController paceController = new ThrottlingController(500, 0d);
         Node node = mock(Node.class);
 
         for (int i = 0; i < 2; i++) {
diff --git a/sentinel-extension/sentinel-metric-exporter/src/test/java/com/alibaba/cps/sentinel/metric/exporter/MetricBeanWriterTest.java b/sentinel-extension/sentinel-metric-exporter/src/test/java/com/alibaba/cps/sentinel/metric/exporter/MetricBeanWriterTest.java
new file mode 100644
index 00000000..bcad68d3
--- /dev/null
+++ b/sentinel-extension/sentinel-metric-exporter/src/test/java/com/alibaba/cps/sentinel/metric/exporter/MetricBeanWriterTest.java
@@ -0,0 +1,60 @@
+/*
+ *  Copyright 1999-2022 Alibaba Group Holding Ltd.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+package com.alibaba.cps.sentinel.metric.exporter;
+
+import com.alibaba.csp.sentinel.metric.exporter.jmx.MetricBeanWriter;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * {@link com.alibaba.csp.sentinel.metric.exporter.jmx.MetricBeanWriter} unit test
+ *
+ * @author quguai
+ * @date 2022/12/7 21:33
+ */
+public class MetricBeanWriterTest {
+
+    @Test
+    public void testEscapeSpecialCharacter() {
+        String character = MetricBeanWriter.escapeSpecialCharacter(null);
+        Assert.assertNull(character);
+
+        character = MetricBeanWriter.escapeSpecialCharacter("");
+        Assert.assertEquals("", character);
+
+        character = MetricBeanWriter.escapeSpecialCharacter("sentinel");
+        Assert.assertEquals("sentinel", character);
+
+        character = MetricBeanWriter.escapeSpecialCharacter("*sentinel");
+        Assert.assertEquals("\"\\*sentinel\"", character);
+
+        character = MetricBeanWriter.escapeSpecialCharacter("?sentinel");
+        Assert.assertEquals("\"\\?sentinel\"", character);
+
+        character = MetricBeanWriter.escapeSpecialCharacter("\nsentinel");
+        Assert.assertEquals("\"\\nsentinel\"", character);
+
+        character = MetricBeanWriter.escapeSpecialCharacter("\"sentinel");
+        Assert.assertEquals("\"\\\"sentinel\"", character);
+
+        character = MetricBeanWriter.escapeSpecialCharacter("=sentinel");
+        Assert.assertEquals("\"=sentinel\"", character);
+
+        character = MetricBeanWriter.escapeSpecialCharacter(":sentinel");
+        Assert.assertEquals("\":sentinel\"", character);
+    }
+}