Fix concurrency limiter issue

jd-opensource · Feb 5, 2025 · 83ef8ef · 83ef8ef
1 parent 064d386
commit 83ef8ef
Show file tree

Hide file tree

Showing 5 changed files with 22 additions and 3 deletions.
diff --git a/...java/com/jd/live/agent/governance/invoke/concurrencylimit/AbstractConcurrencyLimiter.java b/...java/com/jd/live/agent/governance/invoke/concurrencylimit/AbstractConcurrencyLimiter.java
@@ -30,4 +30,9 @@ public boolean acquire() {
      * @return true if the acquisition is successful, false otherwise.
      */
     protected abstract boolean doAcquire();
+
+    @Override
+    public void complete() {
+    }
+
 }
diff --git a/...rc/main/java/com/jd/live/agent/governance/invoke/concurrencylimit/ConcurrencyLimiter.java b/...rc/main/java/com/jd/live/agent/governance/invoke/concurrencylimit/ConcurrencyLimiter.java
@@ -25,5 +25,5 @@
  */
 public interface ConcurrencyLimiter extends Licensee<ConcurrencyLimitPolicy> {
 
-
+    void complete();
 }
diff --git a/.../main/java/com/jd/live/agent/governance/invoke/filter/inbound/ConcurrencyLimitFilter.java b/.../main/java/com/jd/live/agent/governance/invoke/filter/inbound/ConcurrencyLimitFilter.java
@@ -31,6 +31,7 @@
 import com.jd.live.agent.governance.policy.service.limit.ConcurrencyLimitPolicy;
 import com.jd.live.agent.governance.request.ServiceRequest.InboundRequest;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletionStage;
@@ -65,6 +66,7 @@ public void initialize() {
     public <T extends InboundRequest> CompletionStage<Object> filter(InboundInvocation<T> invocation, InboundFilterChain chain) {
         ServicePolicy servicePolicy = invocation.getServiceMetadata().getServicePolicy();
         List<ConcurrencyLimitPolicy> concurrencyLimitPolicies = servicePolicy == null ? null : servicePolicy.getConcurrencyLimitPolicies();
+        List<ConcurrencyLimiter> limiters = new ArrayList<>();
         if (null != concurrencyLimitPolicies && !concurrencyLimitPolicies.isEmpty()) {
             for (ConcurrencyLimitPolicy policy : concurrencyLimitPolicies) {
                 // match logic
@@ -75,10 +77,17 @@ public <T extends InboundRequest> CompletionStage<Object> filter(InboundInvocati
                                 "The request is rejected by concurrency limiter. maxConcurrency=" +
                                         policy.getMaxConcurrency());
                     }
+                    if (limiter != null) {
+                        limiters.add(limiter);
+                    }
                 }
             }
         }
-        return chain.filter(invocation);
+        return chain.filter(invocation).whenComplete((o, throwable) -> {
+            for (ConcurrencyLimiter limiter : limiters) {
+                limiter.complete();
+            }
+        });
     }
 
     /**

diff --git a/...t/implement/flowcontrol/concurrencylimit/resilience4j/Resilience4jConcurrencyLimiter.java b/...t/implement/flowcontrol/concurrencylimit/resilience4j/Resilience4jConcurrencyLimiter.java
@@ -50,4 +50,9 @@ public Resilience4jConcurrencyLimiter(ConcurrencyLimitPolicy policy) {
     public boolean doAcquire() {
         return bulkhead.tryAcquirePermission();
     }
+
+    @Override
+    public void complete() {
+        bulkhead.onComplete();
+    }
 }
diff --git a/...live/agent/plugin/transmission/springweb/v5/definition/BodyInserterRequestDefinition.java b/...live/agent/plugin/transmission/springweb/v5/definition/BodyInserterRequestDefinition.java
@@ -34,7 +34,7 @@
  * @since 1.6.0
  */
 @Injectable
-@Extension(value = "BodyInserterRequestDefinition", order = PluginDefinition.ORDER_TRANSMISSION)
+@Extension(value = "BodyInserterRequestDefinition_v5", order = PluginDefinition.ORDER_TRANSMISSION)
 @ConditionalOnTransmissionEnabled
 @ConditionalOnClass(BodyInserterRequestDefinition.TYPE_BODY_INSERTER_QUEST)
 public class BodyInserterRequestDefinition extends PluginDefinitionAdapter {
-Original file line number
+Diff line change
@@ Expand Up / @@ -25,5 +25,5 @@ @@
      */
     public interface ConcurrencyLimiter extends Licensee<ConcurrencyLimitPolicy> {
+        void complete();
     }