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 db3f8bb4fd fix(ai): return the original compressed bytes when response
transformation falls back (#7920)
db3f8bb4fd is described below
commit db3f8bb4fd28f534e96430997c401ef25931d3fe
Author: hanzhe-one <[email protected]>
AuthorDate: Fri Oct 23 11:24:55 2026 +0800
fix(ai): return the original compressed bytes when response transformation
falls back (#7920)
The response transformer decompressed a gzip upstream body into a String and
used that decoded representation both to build the prompt and, when the
model
failed or produced no usable body, as the response body. The response
headers
were left untouched, so the client received plaintext that still declared
`Content-Encoding: gzip` and a Content-Length of the decoded text. A
gzip-decoding client cannot read that body.
Keep the decoded text for the prompt only, and write the original wire bytes
back on every fallback path:
- model error: write the original bytes instead of the decoded String
- invalid or missing generated body: write the original bytes
- a successful transformation always writes plain UTF-8, so drop an
inherited
Content-Encoding instead of letting the upstream encoding describe new
bytes
Fixes #7908
---
.../response/AiResponseTransformerPlugin.java | 48 +++++---
.../response/AiResponseTransformerPluginTest.java | 129 ++++++++++++++++++++-
2 files changed, 158 insertions(+), 19 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-response-transformer/src/main/java/org/apache/shenyu/plugin/ai/transformer/response/AiResponseTransformerPlugin.java
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-response-transformer/src/main/java/org/apache/shenyu/plugin/ai/transformer/response/AiResponseTransformerPlugin.java
index 97aea95529..1fcd2d0581 100644
---
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-response-transformer/src/main/java/org/apache/shenyu/plugin/ai/transformer/response/AiResponseTransformerPlugin.java
+++
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-response-transformer/src/main/java/org/apache/shenyu/plugin/ai/transformer/response/AiResponseTransformerPlugin.java
@@ -367,7 +367,9 @@ public class AiResponseTransformerPlugin extends
AbstractShenyuPlugin {
byte[] bytes = new byte[dataBuffer.readableByteCount()];
dataBuffer.read(bytes);
DataBufferUtils.release(dataBuffer);
- String originalResponseBody = new String(bytes,
StandardCharsets.UTF_8);
+ // the wire bytes are kept untouched, so a fallback can return
exactly what the upstream sent
+ final byte[] originalWireBytes = bytes;
+ String promptResponseBody = new String(bytes,
StandardCharsets.UTF_8);
String contentEncoding =
exchange.getResponse().getHeaders().getFirst("Content-Encoding");
if ("gzip".equalsIgnoreCase(contentEncoding)) {
@@ -379,15 +381,16 @@ public class AiResponseTransformerPlugin extends
AbstractShenyuPlugin {
while ((len = gis.read(buffer)) > 0) {
bos.write(buffer, 0, len);
}
- originalResponseBody =
bos.toString(StandardCharsets.UTF_8.name());
- LOG.debug("Decompressed response body: {}",
originalResponseBody);
+ promptResponseBody =
bos.toString(StandardCharsets.UTF_8.name());
+ LOG.debug("Decompressed response body: {}",
promptResponseBody);
} catch (Exception e) {
LOG.error("Failed to decompress gzip response", e);
}
}
-
- final String finalResponseBody = originalResponseBody;
-
+
+ // the decoded representation is only used to build the
prompt, never as a response body
+ final String finalPromptResponseBody = promptResponseBody;
+
return aiResponseTransformerTemplate.assembleMessage(exchange)
.flatMap(message -> {
@@ -397,26 +400,25 @@ public class AiResponseTransformerPlugin extends
AbstractShenyuPlugin {
if (messageNode.has("response") &&
messageNode.get("response").isObject()) {
ObjectNode responseNode = (ObjectNode)
messageNode.get("response");
- responseNode.put("body",
finalResponseBody);
+ responseNode.put("body",
finalPromptResponseBody);
}
-
+
messageWithResponseBody =
MAPPER.writeValueAsString(messageNode);
} catch (Exception e) {
LOG.error("Failed to update message with
response body", e);
- messageWithResponseBody =
message.replace("\"body\":\"\"", "\"body\":\"" +
finalResponseBody.replace("\"", "\\\"") + "\"");
+ messageWithResponseBody =
message.replace("\"body\":\"\"", "\"body\":\"" +
finalPromptResponseBody.replace("\"", "\\\"") + "\"");
LOG.info("Fallback message with response body:
{}", messageWithResponseBody);
}
-
+
final String finalMessage =
messageWithResponseBody;
return
chatClient.prompt().user(finalMessage).stream().content()
.filter(Objects::nonNull)
.collect(StringBuilder::new,
StringBuilder::append)
.map(StringBuilder::toString)
- .flatMap(aiResponse ->
applyTransformedResponse(aiResponse, finalResponseBody))
+ .flatMap(aiResponse ->
applyTransformedResponse(aiResponse, originalWireBytes))
.onErrorResume(throwable -> {
LOG.error("response transformation
failed", throwable);
- LOG.info("Returning original response
due to error: {}", finalResponseBody);
- return
WebFluxResultUtils.result(this.exchange,
finalResponseBody.getBytes(StandardCharsets.UTF_8));
+ return
writeOriginalResponse(originalWireBytes);
});
});
}).defaultIfEmpty(Mono.defer(() -> super.writeWith(Flux.empty())))
@@ -429,7 +431,7 @@ public class AiResponseTransformerPlugin extends
AbstractShenyuPlugin {
return writeWith(Flux.from(body).concatMap(Flux::from));
}
- private Mono<Void> applyTransformedResponse(final String aiResponse,
final String originalResponseBody) {
+ private Mono<Void> applyTransformedResponse(final String aiResponse,
final byte[] originalWireBytes) {
HttpHeaders newHeaders = extractHeadersFromAiResponse(aiResponse);
String newBody = extractBodyFromAiResponse(aiResponse);
Integer newStatus = extractStatusFromAiResponse(aiResponse);
@@ -448,6 +450,8 @@ public class AiResponseTransformerPlugin extends
AbstractShenyuPlugin {
if (Objects.nonNull(newBody) && !newBody.isEmpty()) {
this.getHeaders().clear();
this.getHeaders().putAll(newHeaders);
+ // the body is always written as plain UTF-8, so an inherited
encoding header would misdescribe it
+ this.getHeaders().remove(HttpHeaders.CONTENT_ENCODING);
if (Objects.nonNull(newStatus)) {
this.setStatusCode(HttpStatus.valueOf(newStatus));
}
@@ -455,8 +459,20 @@ public class AiResponseTransformerPlugin extends
AbstractShenyuPlugin {
return WebFluxResultUtils.result(this.exchange,
newBody.getBytes(StandardCharsets.UTF_8));
}
// the generated response has no usable body, so the original
response (and its status) is kept
- LOG.warn("response body is empty or null, returning original
response: {}", originalResponseBody);
- return WebFluxResultUtils.result(this.exchange,
originalResponseBody.getBytes(StandardCharsets.UTF_8));
+ LOG.warn("response body is empty or null, returning original
response");
+ return writeOriginalResponse(originalWireBytes);
+ }
+
+ /**
+ * Write the upstream response back unchanged, headers included.
+ * The bytes are the original wire representation, so the upstream
{@code Content-Encoding}
+ * and {@code Content-Length} still describe them.
+ *
+ * @param originalWireBytes the bytes as received from the upstream
+ * @return the completion signal
+ */
+ private Mono<Void> writeOriginalResponse(final byte[]
originalWireBytes) {
+ return
super.writeWith(Mono.just(this.bufferFactory().wrap(originalWireBytes)));
}
private String extractBodyFromAiResponse(final String aiResponse) {
diff --git
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-response-transformer/src/test/java/org/apache/shenyu/plugin/ai/transformer/response/AiResponseTransformerPluginTest.java
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-response-transformer/src/test/java/org/apache/shenyu/plugin/ai/transformer/response/AiResponseTransformerPluginTest.java
index 7f68bdd52d..66e6da9b21 100644
---
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-response-transformer/src/test/java/org/apache/shenyu/plugin/ai/transformer/response/AiResponseTransformerPluginTest.java
+++
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-response-transformer/src/test/java/org/apache/shenyu/plugin/ai/transformer/response/AiResponseTransformerPluginTest.java
@@ -43,6 +43,7 @@ import
org.springframework.context.ConfigurableApplicationContext;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.ai.chat.model.ChatModel;
import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
@@ -54,13 +55,17 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
import com.fasterxml.jackson.databind.ObjectMapper;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -350,9 +355,6 @@ class AiResponseTransformerPluginTest {
ChatClient chatClient = mock(ChatClient.class, RETURNS_DEEP_STUBS);
when(chatClient.prompt().user(anyString()).stream().content())
.thenReturn(Flux.just("HTTP/1.1 200 OK\nX-New: transformed"));
- ConfigurableApplicationContext context =
mock(ConfigurableApplicationContext.class);
- when(context.getBean(ShenyuResult.class)).thenReturn(new
DefaultShenyuResult());
- SpringBeanUtils.getInstance().setApplicationContext(context);
AiResponseTransformerPlugin.AiResponseTransformerDecorator decorator =
new
AiResponseTransformerPlugin.AiResponseTransformerDecorator(exchange, template,
chatClient);
DataBuffer responseBody =
response.bufferFactory().wrap("original".getBytes(StandardCharsets.UTF_8));
@@ -361,6 +363,127 @@ class AiResponseTransformerPluginTest {
assertEquals("original", response.getHeaders().getFirst("X-Original"));
assertNull(response.getHeaders().getFirst("X-New"));
+ assertEquals("original", response.getBodyAsString().block());
+ }
+
+ @Test
+ void testGzipFallbackOnModelErrorKeepsOriginalCompressedBytes() throws
IOException {
+ MockServerHttpResponse response = (MockServerHttpResponse)
exchange.getResponse();
+ response.getHeaders().set(HttpHeaders.CONTENT_ENCODING, "gzip");
+ AiResponseTransformerTemplate template =
mock(AiResponseTransformerTemplate.class);
+
when(template.assembleMessage(exchange)).thenReturn(Mono.just("{\"response\":{\"body\":\"\"}}"));
+ ChatClient chatClient = mock(ChatClient.class, RETURNS_DEEP_STUBS);
+ when(chatClient.prompt().user(anyString()).stream().content())
+ .thenReturn(Flux.error(new IllegalStateException("model
unavailable")));
+ registerResultBean();
+ AiResponseTransformerPlugin.AiResponseTransformerDecorator decorator =
+ new
AiResponseTransformerPlugin.AiResponseTransformerDecorator(exchange, template,
chatClient);
+ byte[] gzipBytes = gzip("{\"customer\":\"alice\"}");
+
+
StepVerifier.create(decorator.writeWith(Mono.just(response.bufferFactory().wrap(gzipBytes))))
+ .verifyComplete();
+
+ assertEquals(HttpStatus.OK, response.getStatusCode());
+ assertEquals("gzip",
response.getHeaders().getFirst(HttpHeaders.CONTENT_ENCODING));
+ assertArrayEquals(gzipBytes, bodyBytes(response));
+ assertEquals("{\"customer\":\"alice\"}", gunzip(bodyBytes(response)));
+ }
+
+ @Test
+ void testGzipFallbackOnInvalidGeneratedBodyKeepsOriginalCompressedBytes()
throws IOException {
+ MockServerHttpResponse response = (MockServerHttpResponse)
exchange.getResponse();
+ response.getHeaders().set(HttpHeaders.CONTENT_ENCODING, "gzip");
+ AiResponseTransformerTemplate template =
mock(AiResponseTransformerTemplate.class);
+
when(template.assembleMessage(exchange)).thenReturn(Mono.just("{\"response\":{\"body\":\"\"}}"));
+ ChatClient chatClient = mock(ChatClient.class, RETURNS_DEEP_STUBS);
+ when(chatClient.prompt().user(anyString()).stream().content())
+ .thenReturn(Flux.just("I cannot transform this response."));
+ registerResultBean();
+ AiResponseTransformerPlugin.AiResponseTransformerDecorator decorator =
+ new
AiResponseTransformerPlugin.AiResponseTransformerDecorator(exchange, template,
chatClient);
+ byte[] gzipBytes = gzip("{\"customer\":\"alice\"}");
+
+
StepVerifier.create(decorator.writeWith(Mono.just(response.bufferFactory().wrap(gzipBytes))))
+ .verifyComplete();
+
+ assertEquals(HttpStatus.OK, response.getStatusCode());
+ assertEquals("gzip",
response.getHeaders().getFirst(HttpHeaders.CONTENT_ENCODING));
+ assertArrayEquals(gzipBytes, bodyBytes(response));
+ assertEquals("{\"customer\":\"alice\"}", gunzip(bodyBytes(response)));
+ }
+
+ @Test
+ void testUncompressedFallbackOnModelErrorKeepsOriginalBytes() {
+ AiResponseTransformerTemplate template =
mock(AiResponseTransformerTemplate.class);
+
when(template.assembleMessage(exchange)).thenReturn(Mono.just("{\"response\":{\"body\":\"\"}}"));
+ ChatClient chatClient = mock(ChatClient.class, RETURNS_DEEP_STUBS);
+ when(chatClient.prompt().user(anyString()).stream().content())
+ .thenReturn(Flux.error(new IllegalStateException("model
unavailable")));
+ registerResultBean();
+ AiResponseTransformerPlugin.AiResponseTransformerDecorator decorator =
+ new
AiResponseTransformerPlugin.AiResponseTransformerDecorator(exchange, template,
chatClient);
+ MockServerHttpResponse response = (MockServerHttpResponse)
exchange.getResponse();
+ byte[] originalBytes =
"{\"customer\":\"alice\"}".getBytes(StandardCharsets.UTF_8);
+
+
StepVerifier.create(decorator.writeWith(Mono.just(response.bufferFactory().wrap(originalBytes))))
+ .verifyComplete();
+
+ assertEquals(HttpStatus.OK, response.getStatusCode());
+
assertNull(response.getHeaders().getFirst(HttpHeaders.CONTENT_ENCODING));
+ assertArrayEquals(originalBytes, bodyBytes(response));
+ }
+
+ @Test
+ void testGzipUpstreamTransformedBodyDropsEncodingHeader() throws
IOException {
+ MockServerHttpResponse response = (MockServerHttpResponse)
exchange.getResponse();
+ response.getHeaders().set(HttpHeaders.CONTENT_ENCODING, "gzip");
+ AiResponseTransformerTemplate template =
mock(AiResponseTransformerTemplate.class);
+
when(template.assembleMessage(exchange)).thenReturn(Mono.just("{\"response\":{\"body\":\"\"}}"));
+ ChatClient chatClient = mock(ChatClient.class, RETURNS_DEEP_STUBS);
+ // the model echoes the upstream encoding header even though its body
is plain text
+ when(chatClient.prompt().user(anyString()).stream().content())
+ .thenReturn(Flux.just("HTTP/1.1 200 OK\nContent-Type:
application/json\nContent-Encoding: gzip\n\n{\"customer\":\"alice\"}"));
+ ConfigurableApplicationContext context =
mock(ConfigurableApplicationContext.class);
+ when(context.getBean(ShenyuResult.class)).thenReturn(new
DefaultShenyuResult());
+ SpringBeanUtils.getInstance().setApplicationContext(context);
+ AiResponseTransformerPlugin.AiResponseTransformerDecorator decorator =
+ new
AiResponseTransformerPlugin.AiResponseTransformerDecorator(exchange, template,
chatClient);
+
+
StepVerifier.create(decorator.writeWith(Mono.just(response.bufferFactory().wrap(gzip("{\"customer\":\"alice\"}")))))
+ .verifyComplete();
+
+ assertEquals(HttpStatus.OK, response.getStatusCode());
+
assertNull(response.getHeaders().getFirst(HttpHeaders.CONTENT_ENCODING));
+ assertEquals("{\"customer\":\"alice\"}", new
String(bodyBytes(response), StandardCharsets.UTF_8));
+ }
+
+ private static void registerResultBean() {
+ ConfigurableApplicationContext context =
mock(ConfigurableApplicationContext.class);
+ lenient().when(context.getBean(ShenyuResult.class)).thenReturn(new
DefaultShenyuResult());
+ SpringBeanUtils.getInstance().setApplicationContext(context);
+ }
+
+ private static byte[] gzip(final String value) throws IOException {
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ try (GZIPOutputStream gos = new GZIPOutputStream(bos)) {
+ gos.write(value.getBytes(StandardCharsets.UTF_8));
+ }
+ return bos.toByteArray();
+ }
+
+ private static String gunzip(final byte[] bytes) throws IOException {
+ try (GZIPInputStream gis = new GZIPInputStream(new
ByteArrayInputStream(bytes))) {
+ return new String(gis.readAllBytes(), StandardCharsets.UTF_8);
+ }
+ }
+
+ private static byte[] bodyBytes(final MockServerHttpResponse response) {
+ return DataBufferUtils.join(response.getBody()).map(dataBuffer -> {
+ byte[] bytes = new byte[dataBuffer.readableByteCount()];
+ dataBuffer.read(bytes);
+ DataBufferUtils.release(dataBuffer);
+ return bytes;
+ }).block();
}
@Test
