[type:fix] release the clients that are replaced or removed (#7163)

shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-token-limiter/src/main/java/org/apache/shenyu/plugin/ai/token/limiter/handler/AiTokenLimiterPluginHandler.java

@@ -78,6 +78,18 @@ public void handlerPlugin(final PluginData pluginData) {
         }
     }
 
+    @Override
+    public void removePlugin(final PluginData pluginData) {
+        final ReactiveRedisTemplate redisTemplate = REDIS_CACHED_HANDLE.get()
+                .obtainHandle(PluginEnum.AI_TOKEN_LIMITER.getName());
+        if (Objects.nonNull(redisTemplate)) {
+            // the client is not used any more, its connection pool and its threads must not stay alive
+            RedisConnectionFactory.destroyQuietly(redisTemplate.getConnectionFactory());
+        }
+        REDIS_CACHED_HANDLE.get().removeHandle(PluginEnum.AI_TOKEN_LIMITER.getName());
+        REDIS_PROPERTIES_CACHED_HANDLE.get().removeHandle(PluginEnum.AI_TOKEN_LIMITER.getName());
+    }
+
     @Override
     public void handlerSelector(final SelectorData selectorData) {
         if (!selectorData.getContinued()) {

shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-token-limiter/src/test/java/org/apache/shenyu/plugin/ai/token/limiter/handler/AiTokenLimiterPluginHandlerTest.java

@@ -84,6 +84,18 @@ public void testHandlerPluginDisabledDoesNothing() {
         assertNull(redisTemplate());
     }
 
+    @Test
+    public void testRemovePluginReleasesTheClient() {
+        AiTokenLimiterPluginHandler handler = new AiTokenLimiterPluginHandler();
+        handler.handlerPlugin(pluginData("127.0.0.1:6379"));
+        ReactiveRedisTemplate<?, ?> cached = redisTemplate();
+        assertNotNull(cached);
+
+        handler.removePlugin(new PluginData());
+        assertFalse(lettuceFactory(cached).isRunning());
+        assertNull(redisTemplate());
+    }
+
     private ReactiveRedisTemplate<?, ?> redisTemplate() {
         return AiTokenLimiterPluginHandler.REDIS_CACHED_HANDLE.get()
                 .obtainHandle(PluginEnum.AI_TOKEN_LIMITER.getName());

shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/handler/CachePluginDataHandler.java

@@ -77,9 +77,13 @@ public void handlerPlugin(final PluginData pluginData) {
             return;
         }
         Singleton.INST.single(CacheConfig.class, cacheConfig);
-        this.closeCacheIfNeed();
         final ICacheBuilder cacheBuilder = ExtensionLoader.getExtensionLoader(ICacheBuilder.class).getJoin(cacheConfig.getCacheType());
+        final ICache lastCache = CacheUtils.getCache();
+        ApplicationConfigCache.getInstance().invalidateAll();
+        // install the new cache before closing the previous one: once a cache closes, its client is
+        // released, and a request that is handed that cache fails.
         Singleton.INST.single(ICache.class, cacheBuilder.builderCache(config));
+        this.closeCache(lastCache);
     }
 
     @Override
@@ -129,10 +133,19 @@ public String pluginNamed() {
     private void closeCacheIfNeed() {
         ICache lastCache = CacheUtils.getCache();
         ApplicationConfigCache.getInstance().invalidateAll();
-        if (Objects.nonNull(lastCache)) {
+        this.closeCache(lastCache);
+    }
+
+    /**
+     * close the given cache, if it exists.
+     *
+     * @param cache the cache to close, may be null
+     */
+    private void closeCache(final ICache cache) {
+        if (Objects.nonNull(cache)) {
             // close last cache.
-            LOG.info("close the last cache {}", lastCache);
-            lastCache.close();
+            LOG.info("close the last cache {}", cache);
+            cache.close();
         }
     }
 }

shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/test/java/org/apache/shenyu/plugin/cache/CachePluginDataHandlerReplacementTest.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.shenyu.plugin.cache;
+
+import org.apache.shenyu.common.dto.PluginData;
+import org.apache.shenyu.common.utils.Singleton;
+import org.apache.shenyu.plugin.cache.handler.CachePluginDataHandler;
+import org.apache.shenyu.plugin.cache.utils.CacheUtils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Test cases for the cache replacement of {@link CachePluginDataHandler}.
+ *
+ * <p>They deliberately do not use the embedded redis of {@code CachePluginDataHandlerTest}: the memory
+ * cache is enough to observe the order in which the previous cache is closed and the new one is
+ * installed.
+ */
+public class CachePluginDataHandlerReplacementTest {
+
+    @Test
+    public void handlerPluginInstallsTheNewCacheBeforeClosingThePreviousOne() {
+        // stand in for the cache that a configuration change replaces
+        ICache previousCache = Mockito.mock(ICache.class);
+        AtomicReference<ICache> cacheWhileClosing = new AtomicReference<>();
+        Mockito.doAnswer(invocation -> {
+            cacheWhileClosing.set(CacheUtils.getCache());
+            return null;
+        }).when(previousCache).close();
+        Singleton.INST.single(ICache.class, previousCache);
+
+        final PluginData pluginData = new PluginData();
+        pluginData.setEnabled(true);
+        // a config that differs from whatever another test left in the singleton
+        pluginData.setConfig("{\"cacheType\":\"memory\",\"probe\":\"" + UUID.randomUUID() + "\"}");
+
+        new CachePluginDataHandler().handlerPlugin(pluginData);
+
+        Mockito.verify(previousCache).close();
+        // a cache must never be handed out once its client has been released
+        Assertions.assertNotNull(cacheWhileClosing.get());
+        Assertions.assertNotSame(previousCache, cacheWhileClosing.get());
+    }
+}