This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new cfa35223ce fix(cache): release joined response buffer on cache misses
(#7481)
cfa35223ce is described below
commit cfa35223ce172a68985e0abbe004142c20bd80ee
Author: lymerin <[email protected]>
AuthorDate: Thu Oct 8 14:35:09 2026 +0800
fix(cache): release joined response buffer on cache misses (#7481)
---
.../apache/shenyu/plugin/cache/CachePlugin.java | 6 ++-
.../shenyu/plugin/cache/CachePluginTest.java | 59 +++++++++++++++++-----
2 files changed, 50 insertions(+), 15 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/CachePlugin.java
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/CachePlugin.java
index 2bb5378ddb..d89fe42789 100644
---
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/CachePlugin.java
+++
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/main/java/org/apache/shenyu/plugin/cache/CachePlugin.java
@@ -130,7 +130,11 @@ public class CachePlugin extends AbstractShenyuPlugin {
final Mono<DataBuffer> dataBufferMono = DataBufferUtils.join(body);
return dataBufferMono.flatMap(dataBuffer -> {
byte[] bytes = new byte[dataBuffer.readableByteCount()];
- dataBuffer.read(bytes);
+ try {
+ dataBuffer.read(bytes);
+ } finally {
+ DataBufferUtils.release(dataBuffer);
+ }
return WebFluxResultUtils.result(this.exchange,
cacheResponse(bytes));
});
}
diff --git
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/test/java/org/apache/shenyu/plugin/cache/CachePluginTest.java
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/test/java/org/apache/shenyu/plugin/cache/CachePluginTest.java
index 891f1cdfa6..e2198a0414 100644
---
a/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/test/java/org/apache/shenyu/plugin/cache/CachePluginTest.java
+++
b/shenyu-plugin/shenyu-plugin-cache/shenyu-plugin-cache-handler/src/test/java/org/apache/shenyu/plugin/cache/CachePluginTest.java
@@ -17,6 +17,7 @@
package org.apache.shenyu.plugin.cache;
+import io.netty.buffer.PooledByteBufAllocator;
import org.apache.shenyu.common.dto.RuleData;
import org.apache.shenyu.common.dto.SelectorData;
import org.apache.shenyu.common.dto.convert.rule.impl.CacheRuleHandle;
@@ -32,14 +33,18 @@ import org.apache.shenyu.plugin.cache.memory.MemoryCache;
import org.apache.shenyu.plugin.cache.utils.CacheUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.http.HttpStatus;
+import org.springframework.core.io.buffer.NettyDataBuffer;
+import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.http.MediaType;
-import org.springframework.mock.http.client.reactive.MockClientHttpResponse;
import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
import org.springframework.mock.web.server.MockServerWebExchange;
import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;
@@ -74,20 +79,46 @@ public class CachePluginTest {
Assertions.assertEquals(cachePlugin.named(),
PluginEnum.CACHE.getName());
}
- @Test
- public void httpResponseTest() {
+ @ParameterizedTest
+ @CsvSource({"1, false", "2, false", "1, true"})
+ public void testCacheMissReleasesJoinedBufferOnce(final int chunks, final
boolean retainExtraReference) {
ConfigurableApplicationContext context =
mock(ConfigurableApplicationContext.class);
- final DefaultShenyuResult shenyuResult = new DefaultShenyuResult();
- when(context.getBean(ShenyuResult.class)).thenReturn(shenyuResult);
+ when(context.getBean(ShenyuResult.class)).thenReturn(new
DefaultShenyuResult());
SpringBeanUtils.getInstance().setApplicationContext(context);
- ServerWebExchange exchange =
MockServerWebExchange.from(MockServerHttpRequest.get("localhost").build());
- MockClientHttpResponse clientResponse = new
MockClientHttpResponse(HttpStatus.OK);
- clientResponse.setBody("body");
- final CacheRuleHandle cacheRuleHandle = new CacheRuleHandle();
- CachePlugin.CacheHttpResponse cacheHttpResponse = new
CachePlugin.CacheHttpResponse(exchange, cacheRuleHandle, "");
- cacheHttpResponse.getHeaders().add("Content-Type",
MediaType.APPLICATION_JSON_VALUE);
- final Mono<Void> mono =
cacheHttpResponse.writeWith(clientResponse.getBody());
- StepVerifier.create(mono).expectSubscription().verifyComplete();
+ MockServerWebExchange exchange =
MockServerWebExchange.from(MockServerHttpRequest.get("/buffer-release").build());
+
exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
+ MemoryCache memoryCache = new MemoryCache();
+ NettyDataBufferFactory factory = new
NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT);
+ NettyDataBuffer[] buffers = new NettyDataBuffer[chunks];
+ for (int i = 0; i < chunks; i++) {
+ String content = chunks == 1 ? "body" : (i == 0 ? "bo" : "dy");
+ buffers[i] = factory.allocateBuffer();
+ buffers[i].write(content.getBytes(StandardCharsets.UTF_8));
+ if (retainExtraReference) {
+ buffers[i].retain();
+ }
+ }
+ try (MockedStatic<CacheUtils> cacheUtils =
Mockito.mockStatic(CacheUtils.class, Mockito.CALLS_REAL_METHODS)) {
+ cacheUtils.when(CacheUtils::getCache).thenReturn(memoryCache);
+ CachePlugin.CacheHttpResponse response = new
CachePlugin.CacheHttpResponse(exchange, new CacheRuleHandle(),
"buffer-release");
+ StepVerifier.create(response.writeWith(chunks == 1 ?
Mono.just(buffers[0]) : Flux.fromArray(buffers))).verifyComplete();
+
+ for (NettyDataBuffer buffer : buffers) {
+ Assertions.assertEquals(retainExtraReference ? 1 : 0,
buffer.getNativeBuffer().refCnt());
+ }
+ Assertions.assertEquals("body",
exchange.getResponse().getBodyAsString().block());
+ Assertions.assertEquals(4L,
exchange.getResponse().getHeaders().getContentLength());
+
Assertions.assertArrayEquals("body".getBytes(StandardCharsets.UTF_8),
memoryCache.getData(CacheUtils.dataKey(exchange)).block());
+
Assertions.assertArrayEquals(MediaType.APPLICATION_JSON_VALUE.getBytes(StandardCharsets.UTF_8),
+
memoryCache.getData(CacheUtils.contentTypeKey(exchange)).block());
+ } finally {
+ for (NettyDataBuffer buffer : buffers) {
+ if (buffer.getNativeBuffer().refCnt() > 0) {
+
buffer.getNativeBuffer().release(buffer.getNativeBuffer().refCnt());
+ }
+ }
+ memoryCache.close();
+ }
}
@Test
