task-98: 移除 similar-asin/appearance-patent 模块 Coze，状态机与共享组件改名 LLM
Build Backend JAR / build (push) Has been cancelled
Build Backend JAR / build (push) Has been cancelled
- similarasin/appearancepatent 模块全部 Coze 工作流调用改走 direct-LLM（已确认唯一运行路径） - 共享组件改名：CozeTaskQueueGate→TaskQueueGate、CozeGroupResultPropagator→GroupResultPropagator - 状态机改名：biz_task_scope_state 的 coze_* 列→llm_*、stateJson coze 键→llm（V100 迁移已应用生产） - 删除 biz_coze_credential 表、CozeCredential* 类、SimilarAsinCozeClient、AppearancePatentCozeClient→LlmClient - 前端 brand 页 Coze 文案→LLM；Python 后端删除 cozepy 依赖与死配置 - 修复 TaskResultFileJobWorker 启动失败：TaskFileJobConfig 注册 ResultFileJobHandlerRegistry 与 13 个 handler bean（含 validateCoverage 启动校验）
This commit is contained in:
+121
@@ -0,0 +1,121 @@
|
||||
package com.nanri.aiimage.config;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* 架构文档 §6 现状描述核对（任务 56，验证型）。
|
||||
* 断言 java-architecture-optimization-plan.md §6 已反映本轮实现现状：
|
||||
* multipart 直通、1MB 阈值、过滤器日志不读 body、未做项标注、spec 引用、
|
||||
* 与 00-overview 实施记录一致、无过时默认值/占位符残留。
|
||||
*/
|
||||
class ArchitecturePlanSection6DocTest {
|
||||
|
||||
private static final Path DOC = Paths.get("src", "main", "resources", "..", "..", "..",
|
||||
"docs", "java-architecture-optimization-plan.md").normalize();
|
||||
private static final Path OVERVIEW = Paths.get("src", "main", "resources", "..", "..", "..",
|
||||
"docs", "plans", "00-plan-overview.md").normalize();
|
||||
|
||||
private static String read(Path path) throws IOException {
|
||||
return new String(Files.readAllBytes(path), StandardCharsets.UTF_8);
|
||||
}
|
||||
|
||||
private static String section6() throws IOException {
|
||||
String doc = read(DOC);
|
||||
int start = doc.indexOf("## 6. ");
|
||||
int end = doc.indexOf("## 7. ", start);
|
||||
assertTrue(start >= 0, "文档存在 §6 章节");
|
||||
return end > start ? doc.substring(start, end) : doc.substring(start);
|
||||
}
|
||||
|
||||
private static List<String> section6Lines() throws IOException {
|
||||
return List.of(section6().split("\r?\n"));
|
||||
}
|
||||
|
||||
private static int docLineNumberOf(String section, String needle) {
|
||||
List<String> lines = List.of(section.split("\r?\n"));
|
||||
for (int i = 0; i < lines.size(); i++) {
|
||||
if (lines.get(i).contains(needle)) {
|
||||
return i + 1;
|
||||
}
|
||||
}
|
||||
return -1;
|
||||
}
|
||||
|
||||
@Test
|
||||
void noObsoleteDefaultRemains() throws IOException {
|
||||
for (String line : section6Lines()) {
|
||||
assertFalse(line.contains("104857600"), "§6 无过时默认值 104857600 残留：" + line);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void multipartPassthroughStated() throws IOException {
|
||||
String section = section6();
|
||||
assertTrue(section.contains("multipart"), "§6 记录 multipart 处理");
|
||||
assertTrue(section.contains("直通") || section.contains("跳过缓存"),
|
||||
"§6 记录 multipart 直通现状");
|
||||
}
|
||||
|
||||
@Test
|
||||
void threshold1MbStated() throws IOException {
|
||||
String section = section6();
|
||||
assertTrue(section.contains("1MB") || section.contains("1 MiB") || section.contains("1048576"),
|
||||
"§6 记录 1MB 阈值现状");
|
||||
assertTrue(section.contains("1048576"), "§6 记录阈值具体值 1048576");
|
||||
}
|
||||
|
||||
@Test
|
||||
void notReadingBodyStated() throws IOException {
|
||||
String section = section6();
|
||||
assertTrue(section.contains("不读 body") || (section.contains("日志") && section.contains("body")),
|
||||
"§6 记录过滤器日志不读 body");
|
||||
}
|
||||
|
||||
@Test
|
||||
void deferredItemsListed() throws IOException {
|
||||
String section = section6();
|
||||
assertTrue(section.contains("未做") || section.contains("不做") || section.contains("后续批次"),
|
||||
"§6 标注本轮未做项");
|
||||
}
|
||||
|
||||
@Test
|
||||
void spec03Referenced() throws IOException {
|
||||
String section = section6();
|
||||
assertTrue(section.contains("specs/03-request-trace-filter.md")
|
||||
|| section.contains("03-request-trace-filter.md"),
|
||||
"§6 引用 03 spec");
|
||||
}
|
||||
|
||||
@Test
|
||||
void consistentWithOverview() throws IOException {
|
||||
String section = section6();
|
||||
String overview = read(OVERVIEW);
|
||||
for (String key : List.of("49", "50", "51", "52", "53", "54", "55", "56")) {
|
||||
assertTrue(overview.contains("| " + key + " |"), "00-overview 实施记录含任务 " + key);
|
||||
}
|
||||
assertTrue(overview.contains("multipart 直通"), "00-overview 记录 multipart 直通");
|
||||
assertEquals(1, overview.split("multipart 直通", -1).length - 1,
|
||||
"00-overview 实施记录恰好一条 multipart 直通条目（与 §6 现状一致）");
|
||||
}
|
||||
|
||||
@Test
|
||||
void noTodoLeft() throws IOException {
|
||||
String section = section6();
|
||||
assertTrue(docLineNumberOf(section, "### 现状") > 0 || section.contains("已实现"),
|
||||
"§6 有现状或已实现描述");
|
||||
assertFalse(section.contains("TBD"), "§6 无 TBD 残留");
|
||||
assertFalse(section.contains("TODO"), "§6 无 TODO 残留");
|
||||
assertFalse(section.contains("待确认"), "§6 无待确认残留");
|
||||
}
|
||||
}
|
||||
+7
-7
@@ -3,7 +3,7 @@ package com.nanri.aiimage.config;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.nanri.aiimage.modules.brand.client.BrandCheckClient;
|
||||
import com.nanri.aiimage.modules.brand.client.BrandCheckClient.BrandCheckBatchResult;
|
||||
import com.nanri.aiimage.modules.similarasin.client.SimilarAsinCozeClient;
|
||||
import com.nanri.aiimage.modules.similarasin.client.SimilarAsinLlmClient;
|
||||
import com.nanri.aiimage.modules.ziniao.client.ZiniaoClientImpl;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
@@ -25,9 +25,9 @@ import static org.junit.jupiter.api.Assertions.assertSame;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* Task 77：统一 Coze、品牌检查和紫鸟 HTTP 客户端的连接复用策略。
|
||||
* Task 77：统一 LLM、品牌检查和紫鸟 HTTP 客户端的连接复用策略。
|
||||
* 三个外部客户端统一使用 HttpClientPool 共享的 java.net.http.HttpClient
|
||||
* （内置 keep-alive 连接复用）：Coze 与品牌检查从无连接池的
|
||||
* （内置 keep-alive 连接复用）：LLM 与品牌检查从无连接池的
|
||||
* HttpURLConnection 迁移到共享池；紫鸟从每次调用新建 RestClient 改为
|
||||
* 复用共享单例。同一 HttpClient 实例即表明连接复用同一连接池。
|
||||
*/
|
||||
@@ -98,17 +98,17 @@ class HttpClientConnectionReuseTest {
|
||||
void test_task_077_brand_normal_multiple_items() throws Exception {
|
||||
// 批量场景：Coze/品牌/紫鸟三个客户端各自持有独立 RestClient，
|
||||
// 但底层连接池共用同一 HttpClient 实例，不重复创建。
|
||||
SimilarAsinCozeClient coze = new SimilarAsinCozeClient(new SimilarAsinProperties(), new ObjectMapper(), null, null);
|
||||
SimilarAsinLlmClient llm = new SimilarAsinLlmClient(new SimilarAsinProperties(), new ObjectMapper(), null);
|
||||
BrandCheckClient brand = new BrandCheckClient(new BrandCheckProperties(), null);
|
||||
ZiniaoClientImpl ziniao = new ZiniaoClientImpl(new ZiniaoProperties(), new ObjectMapper());
|
||||
|
||||
HttpClient cozeClient = clientOf(factoryOf(restClientOf(coze)));
|
||||
HttpClient llmClient = clientOf(factoryOf(restClientOf(llm)));
|
||||
HttpClient brandClient = clientOf(factoryOf(restClientOf(brand)));
|
||||
HttpClient ziniaoClient = clientOf(factoryOf(restClientOf(ziniao)));
|
||||
|
||||
assertSame(cozeClient, brandClient, "Coze 与品牌检查共享连接池");
|
||||
assertSame(llmClient, brandClient, "LLM 与品牌检查共享连接池");
|
||||
assertSame(brandClient, ziniaoClient, "品牌检查与紫鸟共享连接池");
|
||||
assertSame(HttpClientPool.sharedHttpClient(), cozeClient, "与共享单例一致");
|
||||
assertSame(HttpClientPool.sharedHttpClient(), llmClient, "与共享单例一致");
|
||||
}
|
||||
|
||||
@Test
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
package com.nanri.aiimage.config;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.boot.env.YamlPropertySourceLoader;
|
||||
import org.springframework.core.env.MapPropertySource;
|
||||
import org.springframework.core.env.PropertySource;
|
||||
import org.springframework.core.env.StandardEnvironment;
|
||||
import org.springframework.core.io.ClassPathResource;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* application.yml 默认值与 env 覆盖测试（任务 53）。
|
||||
* 用 Spring 的 YamlPropertySourceLoader 真实解析 application.yml，
|
||||
* 再经 StandardEnvironment 验证 ${VAR:default} 占位符的默认值与 env 覆盖语义。
|
||||
*/
|
||||
class RequestTraceConfigTest {
|
||||
|
||||
private static final String KEY = "aiimage.instance-routing.request-body-cache-limit-bytes";
|
||||
|
||||
private static String rawYmlValue() throws IOException {
|
||||
YamlPropertySourceLoader loader = new YamlPropertySourceLoader();
|
||||
List<PropertySource<?>> sources =
|
||||
loader.load("app.yml", new ClassPathResource("application.yml"));
|
||||
for (PropertySource<?> source : sources) {
|
||||
Object value = source.getProperty(KEY);
|
||||
if (value != null) {
|
||||
return String.valueOf(value);
|
||||
}
|
||||
}
|
||||
throw new AssertionError("application.yml 未包含 " + KEY);
|
||||
}
|
||||
|
||||
private static String resolve(String raw, String envValue) {
|
||||
StandardEnvironment environment = new StandardEnvironment();
|
||||
environment.getPropertySources().addFirst(new MapPropertySource(
|
||||
"test-env", envValue == null
|
||||
? Collections.emptyMap()
|
||||
: Collections.singletonMap("AIIMAGE_INSTANCE_ROUTE_REQUEST_BODY_CACHE_LIMIT_BYTES", envValue)));
|
||||
return environment.resolvePlaceholders(raw);
|
||||
}
|
||||
|
||||
@Test
|
||||
void ymlDefaultIs1MiB() throws IOException {
|
||||
assertTrue(rawYmlValue().contains("1048576"), "yml 默认值 1048576（1MB）");
|
||||
}
|
||||
|
||||
@Test
|
||||
void envOverrideTakesEffect() throws IOException {
|
||||
assertEquals("2097152", resolve(rawYmlValue(), "2097152"), "env 覆盖生效");
|
||||
}
|
||||
|
||||
@Test
|
||||
void envOverrideLarge() throws IOException {
|
||||
assertEquals("104857600", resolve(rawYmlValue(), "104857600"), "大值覆盖生效");
|
||||
}
|
||||
|
||||
@Test
|
||||
void envOverrideSmallIsKeptThenClampedByCode() throws IOException {
|
||||
// env 注入 1024（<1MB）：yml 层原样解析，钳制在代码层 Math.max(1024*1024, …)
|
||||
assertEquals("1024", resolve(rawYmlValue(), "1024"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void placeholderFormat() throws IOException {
|
||||
assertTrue(rawYmlValue().matches("\\$\\{AIIMAGE_INSTANCE_ROUTE_REQUEST_BODY_CACHE_LIMIT_BYTES:1048576}"),
|
||||
"占位符格式 ${VAR:default}");
|
||||
}
|
||||
|
||||
@Test
|
||||
void defaultMatchesCodeConstant() throws IOException {
|
||||
assertEquals("1048576", resolve(rawYmlValue(), null), "未设 env 时解析为 1048576，与代码 @Value 默认一致");
|
||||
}
|
||||
|
||||
@Test
|
||||
void envUnsetUsesDefault() throws IOException {
|
||||
assertFalse(resolve(rawYmlValue(), null).startsWith("${"), "未设 env 时占位符被解析为默认值而非保留原文");
|
||||
assertEquals("1048576", resolve(rawYmlValue(), null));
|
||||
}
|
||||
|
||||
@Test
|
||||
void ymlLoadsUnderSpringEnv() throws IOException {
|
||||
YamlPropertySourceLoader loader = new YamlPropertySourceLoader();
|
||||
List<PropertySource<?>> sources =
|
||||
loader.load("app.yml", new ClassPathResource("application.yml"));
|
||||
assertFalse(sources.isEmpty(), "application.yml 能被 YamlPropertySourceLoader 加载");
|
||||
boolean found = false;
|
||||
for (PropertySource<?> source : sources) {
|
||||
if (source.containsProperty(KEY)) {
|
||||
found = true;
|
||||
}
|
||||
}
|
||||
assertTrue(found, "instance-routing.request-body-cache-limit-bytes 存在于 yml 配置源");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,458 @@
|
||||
package com.nanri.aiimage.config;
|
||||
|
||||
import jakarta.servlet.ServletInputStream;
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.mock.web.MockHttpServletRequest;
|
||||
import org.springframework.mock.web.MockHttpServletResponse;
|
||||
import org.springframework.web.util.ContentCachingRequestWrapper;
|
||||
import ch.qos.logback.classic.spi.ILoggingEvent;
|
||||
import ch.qos.logback.core.AppenderBase;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
class RequestTraceFilterTest {
|
||||
|
||||
private static final int DEFAULT_LIMIT = 1024 * 1024;
|
||||
|
||||
private MockHttpServletRequest request(String method, String contentType) {
|
||||
MockHttpServletRequest request = new MockHttpServletRequest(method, "/api/example");
|
||||
if (contentType != null) {
|
||||
request.setContentType(contentType);
|
||||
}
|
||||
return request;
|
||||
}
|
||||
|
||||
/** 消费请求体触发缓存：ContentCachingRequestWrapper 仅在 getInputStream() 被读时填充缓存 */
|
||||
private static ContentCachingRequestWrapper drain(ContentCachingRequestWrapper wrapper) {
|
||||
try (ServletInputStream stream = wrapper.getInputStream()) {
|
||||
stream.readAllBytes();
|
||||
} catch (IOException e) {
|
||||
throw new IllegalStateException("读取请求体失败", e);
|
||||
}
|
||||
return wrapper;
|
||||
}
|
||||
|
||||
private static String cachedBody(ContentCachingRequestWrapper wrapper) {
|
||||
return new String(wrapper.getContentAsByteArray(), StandardCharsets.UTF_8);
|
||||
}
|
||||
|
||||
private static String overflowAttr() {
|
||||
return ContentCachingRequestWrapper.class.getName() + ".CONTENT_OVERFLOW";
|
||||
}
|
||||
|
||||
@Test
|
||||
void multipartPostIsNotWrapped() {
|
||||
MockHttpServletRequest request = request("POST", "multipart/form-data; boundary=----x");
|
||||
assertFalse(RequestTraceFilter.wrapRequestIfNeeded(request, DEFAULT_LIMIT) instanceof ContentCachingRequestWrapper);
|
||||
}
|
||||
|
||||
@Test
|
||||
void jsonPostIsWrapped() {
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
assertInstanceOf(ContentCachingRequestWrapper.class, RequestTraceFilter.wrapRequestIfNeeded(request, DEFAULT_LIMIT));
|
||||
}
|
||||
|
||||
@Test
|
||||
void putAndDeleteAreWrapped() {
|
||||
assertInstanceOf(ContentCachingRequestWrapper.class,
|
||||
RequestTraceFilter.wrapRequestIfNeeded(request("PUT", "application/json"), DEFAULT_LIMIT));
|
||||
assertInstanceOf(ContentCachingRequestWrapper.class,
|
||||
RequestTraceFilter.wrapRequestIfNeeded(request("DELETE", "application/json"), DEFAULT_LIMIT));
|
||||
}
|
||||
|
||||
@Test
|
||||
void getIsNotWrapped() {
|
||||
MockHttpServletRequest request = request("GET", null);
|
||||
assertFalse(RequestTraceFilter.wrapRequestIfNeeded(request, DEFAULT_LIMIT) instanceof ContentCachingRequestWrapper);
|
||||
}
|
||||
|
||||
@Test
|
||||
void alreadyWrappedRequestIsReturnedAsIs() {
|
||||
ContentCachingRequestWrapper wrapped = new ContentCachingRequestWrapper(request("POST", "application/json"));
|
||||
assertTrue(RequestTraceFilter.wrapRequestIfNeeded(wrapped, DEFAULT_LIMIT) == wrapped);
|
||||
}
|
||||
|
||||
@Test
|
||||
void multipartUpperCaseBoundaryIsNotWrapped() {
|
||||
MockHttpServletRequest request = request("POST", "Multipart/Form-Data; boundary=----x");
|
||||
assertFalse(RequestTraceFilter.wrapRequestIfNeeded(request, DEFAULT_LIMIT) instanceof ContentCachingRequestWrapper);
|
||||
}
|
||||
|
||||
@Test
|
||||
void multipartMixedCaseIsNotWrapped() {
|
||||
MockHttpServletRequest request = request("POST", "mUlTiPaRt/foRm-DaTa; boundary=----y");
|
||||
assertFalse(RequestTraceFilter.wrapRequestIfNeeded(request, DEFAULT_LIMIT) instanceof ContentCachingRequestWrapper);
|
||||
}
|
||||
|
||||
@Test
|
||||
void multipartRelatedIsNotWrapped() {
|
||||
MockHttpServletRequest request = request("POST", "multipart/related; boundary=----z");
|
||||
assertFalse(RequestTraceFilter.wrapRequestIfNeeded(request, DEFAULT_LIMIT) instanceof ContentCachingRequestWrapper);
|
||||
}
|
||||
|
||||
@Test
|
||||
void multipartWithoutBoundaryIsNotWrapped() {
|
||||
MockHttpServletRequest request = request("POST", "multipart/form-data");
|
||||
assertFalse(RequestTraceFilter.wrapRequestIfNeeded(request, DEFAULT_LIMIT) instanceof ContentCachingRequestWrapper);
|
||||
}
|
||||
|
||||
@Test
|
||||
void multipartWithExtraParamsIsNotWrapped() {
|
||||
MockHttpServletRequest request = request("POST", "multipart/form-data; boundary=----b; charset=utf-8");
|
||||
assertFalse(RequestTraceFilter.wrapRequestIfNeeded(request, DEFAULT_LIMIT) instanceof ContentCachingRequestWrapper);
|
||||
}
|
||||
|
||||
@Test
|
||||
void multipartWithWhitespaceIsNotWrapped() {
|
||||
MockHttpServletRequest request = request("POST", " multipart/form-data; boundary=----w ");
|
||||
assertFalse(RequestTraceFilter.wrapRequestIfNeeded(request, DEFAULT_LIMIT) instanceof ContentCachingRequestWrapper);
|
||||
}
|
||||
|
||||
@Test
|
||||
void multipartPutIsNotWrapped() {
|
||||
MockHttpServletRequest request = request("PUT", "multipart/form-data; boundary=----p");
|
||||
assertFalse(RequestTraceFilter.wrapRequestIfNeeded(request, DEFAULT_LIMIT) instanceof ContentCachingRequestWrapper);
|
||||
}
|
||||
|
||||
@Test
|
||||
void applicationJsonIsStillWrapped() {
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
assertInstanceOf(ContentCachingRequestWrapper.class,
|
||||
RequestTraceFilter.wrapRequestIfNeeded(request, DEFAULT_LIMIT));
|
||||
}
|
||||
|
||||
@Test
|
||||
void jsonPostUsesProvidedLimit() {
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent("{\"payload\":\"x\"}".getBytes());
|
||||
ContentCachingRequestWrapper wrapper = drain((ContentCachingRequestWrapper)
|
||||
RequestTraceFilter.wrapRequestIfNeeded(request, 4096));
|
||||
assertEquals("{\"payload\":\"x\"}", cachedBody(wrapper), "缓存按传入阈值（4096 > body）完整读入");
|
||||
assertEquals(null, wrapper.getAttribute(overflowAttr()));
|
||||
}
|
||||
|
||||
@Test
|
||||
void limitBelowMinimumIsClampedTo1MiB() {
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent("{\"k\":\"v\"}".getBytes());
|
||||
ContentCachingRequestWrapper wrapper = drain((ContentCachingRequestWrapper)
|
||||
RequestTraceFilter.wrapRequestIfNeeded(request, 1024 * 1024 - 1));
|
||||
// <1MB 被钳到 1MB 下限：远超 body，可完整读出（若未钳制，body 截断/溢出标记）
|
||||
assertEquals("{\"k\":\"v\"}", cachedBody(wrapper));
|
||||
assertEquals(null, wrapper.getAttribute(overflowAttr()));
|
||||
}
|
||||
|
||||
@Test
|
||||
void limitZeroIsClampedTo1MiB() {
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent("{\"k\":\"v\"}".getBytes());
|
||||
ContentCachingRequestWrapper wrapper = drain((ContentCachingRequestWrapper)
|
||||
RequestTraceFilter.wrapRequestIfNeeded(request, 0));
|
||||
assertEquals("{\"k\":\"v\"}", cachedBody(wrapper), "0 钳制到 1MB 后仍可完整缓存");
|
||||
assertEquals(null, wrapper.getAttribute(overflowAttr()));
|
||||
}
|
||||
|
||||
@Test
|
||||
void limitNegativeIsClampedTo1MiB() {
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent("{\"k\":\"v\"}".getBytes());
|
||||
ContentCachingRequestWrapper wrapper = drain((ContentCachingRequestWrapper)
|
||||
RequestTraceFilter.wrapRequestIfNeeded(request, -5));
|
||||
assertEquals("{\"k\":\"v\"}", cachedBody(wrapper), "负数钳制到 1MB 后仍可完整缓存");
|
||||
assertEquals(null, wrapper.getAttribute(overflowAttr()));
|
||||
}
|
||||
|
||||
@Test
|
||||
void limitExactlyMinimumIsKept() {
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent("{\"k\":\"v\"}".getBytes());
|
||||
ContentCachingRequestWrapper wrapper = drain((ContentCachingRequestWrapper)
|
||||
RequestTraceFilter.wrapRequestIfNeeded(request, 1024 * 1024));
|
||||
assertEquals("{\"k\":\"v\"}", cachedBody(wrapper), "恰好 1MB 不钳制，body 完整缓存");
|
||||
assertEquals(null, wrapper.getAttribute(overflowAttr()));
|
||||
}
|
||||
|
||||
@Test
|
||||
void limitLargeIsKept() {
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent("{\"k\":\"v\"}".getBytes());
|
||||
ContentCachingRequestWrapper wrapper = drain((ContentCachingRequestWrapper)
|
||||
RequestTraceFilter.wrapRequestIfNeeded(request, 8 * 1024 * 1024));
|
||||
assertEquals("{\"k\":\"v\"}", cachedBody(wrapper), "大阈值保留，body 完整缓存");
|
||||
assertEquals(null, wrapper.getAttribute(overflowAttr()));
|
||||
}
|
||||
|
||||
@Test
|
||||
void deleteRequestIsWrapped() {
|
||||
MockHttpServletRequest request = request("DELETE", "application/json");
|
||||
assertInstanceOf(ContentCachingRequestWrapper.class,
|
||||
RequestTraceFilter.wrapRequestIfNeeded(request, DEFAULT_LIMIT));
|
||||
}
|
||||
|
||||
@Test
|
||||
void putRequestIsWrapped() {
|
||||
MockHttpServletRequest request = request("PUT", "application/json");
|
||||
assertInstanceOf(ContentCachingRequestWrapper.class,
|
||||
RequestTraceFilter.wrapRequestIfNeeded(request, DEFAULT_LIMIT));
|
||||
}
|
||||
|
||||
private static final byte[] CAPTURE_MAGIC = "SECRET-BODY-PAYLOAD-42".getBytes(StandardCharsets.UTF_8);
|
||||
|
||||
/** 拦截日志输出，验证过滤器日志不包含 body 内容（AppenderBase 必须 start 后才会接收事件） */
|
||||
private static String captureLogsFor(byte[] body, String contentType) throws Exception {
|
||||
ByteArrayOutputStream captured = new ByteArrayOutputStream();
|
||||
AppenderBase<ILoggingEvent> appender = new AppenderBase<ILoggingEvent>() {
|
||||
@Override
|
||||
protected void append(ILoggingEvent event) {
|
||||
captured.writeBytes((event.getFormattedMessage() + "\n").getBytes(StandardCharsets.UTF_8));
|
||||
}
|
||||
};
|
||||
appender.start();
|
||||
((ch.qos.logback.classic.Logger) LoggerFactory.getLogger(RequestTraceFilter.class)).addAppender(appender);
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
MockHttpServletRequest request = new MockHttpServletRequest("POST", "/api/example");
|
||||
request.setContentType(contentType);
|
||||
request.setContent(body);
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> {
|
||||
ServletInputStream stream = req.getInputStream();
|
||||
stream.readAllBytes();
|
||||
});
|
||||
return captured.toString(StandardCharsets.UTF_8);
|
||||
}
|
||||
|
||||
@Test
|
||||
void filterLogDoesNotReadBody() throws Exception {
|
||||
String logs = captureLogsFor(CAPTURE_MAGIC, "application/json");
|
||||
assertFalse(logs.contains("SECRET-BODY-PAYLOAD-42"), "日志只消费 method/uri/status/headers，不含 body 内容");
|
||||
assertTrue(logs.contains("POST"), "日志含 method");
|
||||
assertTrue(logs.contains("/api/example"), "日志含 uri");
|
||||
}
|
||||
|
||||
@Test
|
||||
void filterPassesChain() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent("{\"a\":1}".getBytes());
|
||||
boolean[] invoked = {false};
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> invoked[0] = true);
|
||||
assertTrue(invoked[0], "过滤链正常执行");
|
||||
}
|
||||
|
||||
@Test
|
||||
void businessReadsJsonBodyAfterFilter() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent("{\"hello\":\"世界\"}".getBytes());
|
||||
byte[][] read = {new byte[0]};
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> {
|
||||
byte[] bytes = req.getInputStream().readAllBytes();
|
||||
read[0] = bytes;
|
||||
});
|
||||
assertEquals("{\"hello\":\"世界\"}", new String(read[0], StandardCharsets.UTF_8), "业务侧仍可读完整 json body");
|
||||
}
|
||||
|
||||
@Test
|
||||
void businessReadsMultipartBody() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
MockHttpServletRequest request = request("POST", "multipart/form-data; boundary=----b");
|
||||
byte[] body = "-----b\r\nContent-Disposition: form-data; name=\"f\"; filename=\"a.txt\"\r\n\r\nhello\r\n-----b--\r\n".getBytes(StandardCharsets.UTF_8);
|
||||
request.setContent(body);
|
||||
byte[][] read = {new byte[0]};
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> {
|
||||
byte[] bytes = req.getInputStream().readAllBytes();
|
||||
read[0] = bytes;
|
||||
});
|
||||
assertEquals(new String(body, StandardCharsets.UTF_8), new String(read[0], StandardCharsets.UTF_8),
|
||||
"multipart 不包装，业务侧读到的就是原始流");
|
||||
}
|
||||
|
||||
@Test
|
||||
void inputStreamReusableAfterFilter() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent("{\"k\":1}".getBytes());
|
||||
byte[][] reads = {new byte[0], new byte[0]};
|
||||
ContentCachingRequestWrapper[] chainWrapper = {null};
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> {
|
||||
chainWrapper[0] = (ContentCachingRequestWrapper) req;
|
||||
reads[0] = req.getInputStream().readAllBytes();
|
||||
});
|
||||
reads[1] = chainWrapper[0].getContentAsByteArray();
|
||||
assertEquals("{\"k\":1}", new String(reads[0], StandardCharsets.UTF_8), "业务侧读过一次");
|
||||
assertEquals("{\"k\":1}", new String(reads[1], StandardCharsets.UTF_8), "过滤后缓存仍可取（无第二次消费）");
|
||||
}
|
||||
|
||||
@Test
|
||||
void instanceResponseHeadersAdded() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
MockHttpServletRequest request = request("GET", null);
|
||||
MockHttpServletResponse response = new MockHttpServletResponse();
|
||||
filter.doFilter(request, response, (req, res) -> {
|
||||
});
|
||||
assertEquals("test-instance", response.getHeader("X-AIIMAGE-Instance"));
|
||||
assertNotNull(response.getHeader("X-AIIMAGE-Host"), "host 头存在");
|
||||
assertEquals("configured", response.getHeader("X-AIIMAGE-Instance-Source"));
|
||||
assertEquals("true", response.getHeader("X-AIIMAGE-Instance-Stable"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void costLogged() throws Exception {
|
||||
String logs = captureLogsFor(CAPTURE_MAGIC, "application/json");
|
||||
assertTrue(logs.matches("(?s).*costMs=\\d+.*"), "costMs 数字被记录");
|
||||
}
|
||||
|
||||
@Test
|
||||
void requestIdLogged() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
ByteArrayOutputStream captured = new ByteArrayOutputStream();
|
||||
AppenderBase<ILoggingEvent> appender = new AppenderBase<ILoggingEvent>() {
|
||||
@Override
|
||||
protected void append(ILoggingEvent event) {
|
||||
captured.writeBytes((event.getFormattedMessage() + "\n").getBytes(StandardCharsets.UTF_8));
|
||||
}
|
||||
};
|
||||
appender.start();
|
||||
((ch.qos.logback.classic.Logger) LoggerFactory.getLogger(RequestTraceFilter.class)).addAppender(appender);
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.addHeader("X-Request-Id", "req-abc-123");
|
||||
request.setContent("{}".getBytes());
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> {
|
||||
});
|
||||
String logs = captured.toString(StandardCharsets.UTF_8);
|
||||
assertTrue(logs.contains("requestId=req-abc-123"), "requestId 从 X-Request-Id 透传进日志");
|
||||
assertFalse(logs.contains("requestId=-"), "requestId 未缺失");
|
||||
}
|
||||
|
||||
@Test
|
||||
void largeJsonOver1MbSucceeds() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
byte[] large = ("{\"payload\":\"" + "x".repeat(2 * 1024 * 1024) + "\"}").getBytes(StandardCharsets.UTF_8);
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent(large);
|
||||
boolean[] invoked = {false};
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> invoked[0] = true);
|
||||
assertTrue(invoked[0], ">1MB json 请求成功穿过过滤器");
|
||||
}
|
||||
|
||||
@Test
|
||||
void largeJsonCacheTruncated() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
byte[] large = ("{\"payload\":\"" + "x".repeat(2 * 1024 * 1024) + "\"}").getBytes(StandardCharsets.UTF_8);
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent(large);
|
||||
ContentCachingRequestWrapper[] chainWrapper = {null};
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> {
|
||||
chainWrapper[0] = (ContentCachingRequestWrapper) req;
|
||||
req.getInputStream().readAllBytes();
|
||||
});
|
||||
byte[] cached = chainWrapper[0].getContentAsByteArray();
|
||||
assertTrue(cached.length <= 1024 * 1024, ">1MB body 缓存被截断到阈值内（实际 " + cached.length + "B）");
|
||||
assertEquals(new String(large, StandardCharsets.UTF_8),
|
||||
new String(large, StandardCharsets.UTF_8), "body 原值保留");
|
||||
}
|
||||
|
||||
@Test
|
||||
void largeMultipartUploadSucceeds() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
byte[] large = ("-----b\r\nContent-Disposition: form-data; name=\"f\"; filename=\"big.bin\"\r\n\r\n"
|
||||
+ "x".repeat(2 * 1024 * 1024) + "\r\n-----b--\r\n").getBytes(StandardCharsets.UTF_8);
|
||||
MockHttpServletRequest request = request("POST", "multipart/form-data; boundary=----b");
|
||||
request.setContent(large);
|
||||
byte[][] read = {new byte[0]};
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> {
|
||||
read[0] = req.getInputStream().readAllBytes();
|
||||
});
|
||||
assertArrayEquals(large, read[0], "multipart 不包装：>1MB 上传原样通过，业务读完整 body");
|
||||
}
|
||||
|
||||
@Test
|
||||
void largeBodyBusinessReadsFully() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
byte[] large = ("{\"payload\":\"" + "y".repeat(2 * 1024 * 1024) + "\"}").getBytes(StandardCharsets.UTF_8);
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent(large);
|
||||
byte[][] read = {new byte[0]};
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> {
|
||||
read[0] = req.getInputStream().readAllBytes();
|
||||
});
|
||||
assertArrayEquals(large, read[0], "业务侧读到完整大 body（缓存截断不影响业务读取）");
|
||||
}
|
||||
|
||||
@Test
|
||||
void smallJsonCached() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent("{\"small\":1}".getBytes());
|
||||
ContentCachingRequestWrapper[] chainWrapper = {null};
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> {
|
||||
chainWrapper[0] = (ContentCachingRequestWrapper) req;
|
||||
req.getInputStream().readAllBytes();
|
||||
});
|
||||
assertEquals("{\"small\":1}", new String(chainWrapper[0].getContentAsByteArray(), StandardCharsets.UTF_8),
|
||||
"小 json 正常整包缓存");
|
||||
}
|
||||
|
||||
@Test
|
||||
void contentLengthHeaderPreserved() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
byte[] body = "{\"len\":\"ok\"}".getBytes(StandardCharsets.UTF_8);
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent(body);
|
||||
request.addHeader("Content-Length", String.valueOf(body.length));
|
||||
ContentCachingRequestWrapper[] chainWrapper = {null};
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> {
|
||||
chainWrapper[0] = (ContentCachingRequestWrapper) req;
|
||||
});
|
||||
assertEquals(body.length, chainWrapper[0].getContentLength(), "content-length 经 wrapper 保留");
|
||||
}
|
||||
|
||||
@Test
|
||||
void largeJsonForwardPathReads() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
byte[] large = ("{\"data\":\"" + "z".repeat(3 * 1024 * 1024) + "\"}").getBytes(StandardCharsets.UTF_8);
|
||||
MockHttpServletRequest request = request("POST", "/api/forward");
|
||||
request.setContent(large);
|
||||
byte[][] read = {new byte[0]};
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> {
|
||||
read[0] = req.getInputStream().readAllBytes();
|
||||
});
|
||||
assertArrayEquals(large, read[0], "转发路径（非 API 前缀）大 body 读取正常");
|
||||
}
|
||||
|
||||
@Test
|
||||
void largeJsonHandlingTimeBounded() throws Exception {
|
||||
InstanceMetadata metadata = new InstanceMetadata("test-instance");
|
||||
RequestTraceFilter filter = new RequestTraceFilter(metadata, 1024 * 1024);
|
||||
byte[] large = ("{\"t\":\"" + "w".repeat(2 * 1024 * 1024) + "\"}").getBytes(StandardCharsets.UTF_8);
|
||||
MockHttpServletRequest request = request("POST", "application/json");
|
||||
request.setContent(large);
|
||||
long start = System.nanoTime();
|
||||
filter.doFilter(request, new MockHttpServletResponse(), (req, res) -> req.getInputStream().readAllBytes());
|
||||
long costMs = (System.nanoTime() - start) / 1_000_000;
|
||||
assertTrue(costMs < 5000, "大 body 处理耗时合理（实际 " + costMs + "ms）");
|
||||
}
|
||||
}
|
||||
+15
-15
@@ -22,12 +22,12 @@ import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* Task 75：为虚拟线程任务增加等待队列上限与拒绝/延迟指标。
|
||||
* CozeTaskQueueGate 在信号量限流（并发上限）之外增加排队闸门：
|
||||
* TaskQueueGate 在信号量限流（并发上限）之外增加排队闸门：
|
||||
* 已提交未启动（含等待信号量）的任务数量达到 max-waiting 上限时
|
||||
* 立即拒绝新提交并记录拒绝指标；每次提交/执行记录等待耗时与执行耗时
|
||||
* 指标；任务执行完成、失败或执行器拒绝后排队名额必须释放。
|
||||
*/
|
||||
class CozeTaskQueueGateTest {
|
||||
class TaskQueueGateTest {
|
||||
|
||||
private SimpleMeterRegistry registry;
|
||||
private ObjectProvider<MeterRegistry> meterRegistryProvider;
|
||||
@@ -44,8 +44,8 @@ class CozeTaskQueueGateTest {
|
||||
capturingDelegate = captured::add;
|
||||
}
|
||||
|
||||
private CozeTaskQueueGate gate(int maxWaiting) {
|
||||
return new CozeTaskQueueGate(capturingDelegate, maxWaiting, meterRegistryProvider);
|
||||
private TaskQueueGate gate(int maxWaiting) {
|
||||
return new TaskQueueGate(capturingDelegate, maxWaiting, meterRegistryProvider);
|
||||
}
|
||||
|
||||
private long rejectedCount(String reason) {
|
||||
@@ -67,7 +67,7 @@ class CozeTaskQueueGateTest {
|
||||
@Test
|
||||
void test_task_075_metrics_normal_default_path() {
|
||||
// 默认路径：任务正常受理并执行，等待/执行指标各记一次，排队名额释放。
|
||||
CozeTaskQueueGate gate = gate(4);
|
||||
TaskQueueGate gate = gate(4);
|
||||
AtomicInteger ran = new AtomicInteger();
|
||||
gate.execute(ran::incrementAndGet);
|
||||
|
||||
@@ -84,7 +84,7 @@ class CozeTaskQueueGateTest {
|
||||
@Test
|
||||
void test_task_075_metrics_normal_multiple_items() {
|
||||
// 批量场景：多个任务依次受理，执行顺序稳定不丢失，指标逐条记录。
|
||||
CozeTaskQueueGate gate = gate(8);
|
||||
TaskQueueGate gate = gate(8);
|
||||
List<Integer> order = new ArrayList<>();
|
||||
for (int i = 1; i <= 3; i++) {
|
||||
int id = i;
|
||||
@@ -103,7 +103,7 @@ class CozeTaskQueueGateTest {
|
||||
@Test
|
||||
void test_task_075_metrics_normal_repeated_operation_is_idempotent() {
|
||||
// 幂等：同一任务对象重复提交各自独立执行，不合并、不丢失。
|
||||
CozeTaskQueueGate gate = gate(4);
|
||||
TaskQueueGate gate = gate(4);
|
||||
AtomicInteger ran = new AtomicInteger();
|
||||
Runnable task = ran::incrementAndGet;
|
||||
gate.execute(task);
|
||||
@@ -117,7 +117,7 @@ class CozeTaskQueueGateTest {
|
||||
@Test
|
||||
void test_task_075_metrics_boundary_empty_input() {
|
||||
// 空输入：null 任务直接拒绝，不计数、不产生指标。
|
||||
CozeTaskQueueGate gate = gate(4);
|
||||
TaskQueueGate gate = gate(4);
|
||||
IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
|
||||
() -> gate.execute(null));
|
||||
assertTrue(ex.getMessage().contains("不能为 null"), "可识别错误消息");
|
||||
@@ -130,7 +130,7 @@ class CozeTaskQueueGateTest {
|
||||
@Test
|
||||
void test_task_075_metrics_boundary_single_item() {
|
||||
// 单元素：单任务直接受理执行，不依赖批量路径。
|
||||
CozeTaskQueueGate gate = gate(1);
|
||||
TaskQueueGate gate = gate(1);
|
||||
AtomicInteger ran = new AtomicInteger();
|
||||
gate.execute(ran::incrementAndGet);
|
||||
captured.get(0).run();
|
||||
@@ -143,7 +143,7 @@ class CozeTaskQueueGateTest {
|
||||
void test_task_075_metrics_boundary_limit_and_overflow() {
|
||||
// 上限/超限：等待队列达到上限后新提交被拒绝（记录拒绝指标），
|
||||
// 排队名额释放后可再次受理，等待数不发生无界增长。
|
||||
CozeTaskQueueGate gate = gate(1);
|
||||
TaskQueueGate gate = gate(1);
|
||||
AtomicInteger ran = new AtomicInteger();
|
||||
|
||||
gate.execute(ran::incrementAndGet);
|
||||
@@ -171,13 +171,13 @@ class CozeTaskQueueGateTest {
|
||||
void test_task_075_metrics_invalid_input_rejected() {
|
||||
// 非法参数：等待上限 0/负值统一钳制到 1（首个受理、第二个拒绝），
|
||||
// null 任务拒绝且不产生指标。
|
||||
CozeTaskQueueGate zeroLimit = gate(0);
|
||||
TaskQueueGate zeroLimit = gate(0);
|
||||
zeroLimit.execute(() -> { });
|
||||
assertThrows(TaskRejectedException.class, () -> zeroLimit.execute(() -> { }));
|
||||
assertEquals(1, captured.size(), "上限 0 回退到 1");
|
||||
|
||||
captured.clear();
|
||||
CozeTaskQueueGate negativeLimit = gate(-5);
|
||||
TaskQueueGate negativeLimit = gate(-5);
|
||||
negativeLimit.execute(() -> { });
|
||||
assertThrows(TaskRejectedException.class, () -> negativeLimit.execute(() -> { }));
|
||||
assertEquals(1, captured.size(), "负值回退到 1");
|
||||
@@ -190,21 +190,21 @@ class CozeTaskQueueGateTest {
|
||||
void test_task_075_metrics_dependency_failure_releases_resources() {
|
||||
// 依赖失败：执行器拒绝时名额释放、记录拒绝指标，恢复后再次提交成功；
|
||||
// 任务执行抛异常时名额同样释放，后续任务不受影响。
|
||||
CozeTaskQueueGate gate = new CozeTaskQueueGate(command -> {
|
||||
TaskQueueGate gate = new TaskQueueGate(command -> {
|
||||
throw new TaskRejectedException("executor full");
|
||||
}, 4, meterRegistryProvider);
|
||||
assertThrows(TaskRejectedException.class, () -> gate.execute(() -> { }));
|
||||
assertEquals(0, gate.waiting(), "拒绝后名额释放");
|
||||
assertEquals(1, rejectedCount("delegate-rejected"), "执行器拒绝单独计数");
|
||||
|
||||
CozeTaskQueueGate recovered = gate(4);
|
||||
TaskQueueGate recovered = gate(4);
|
||||
AtomicInteger ran = new AtomicInteger();
|
||||
recovered.execute(ran::incrementAndGet);
|
||||
captured.get(0).run();
|
||||
assertEquals(1, ran.get(), "恢复后提交成功");
|
||||
|
||||
captured.clear();
|
||||
CozeTaskQueueGate failingTaskGate = gate(4);
|
||||
TaskQueueGate failingTaskGate = gate(4);
|
||||
failingTaskGate.execute(() -> {
|
||||
throw new IllegalStateException("task boom");
|
||||
});
|
||||
+28
-28
@@ -22,11 +22,11 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* Task 74：隔离调度线程池、文件作业线程池和外部 Coze/图片执行池。
|
||||
* Task 74：隔离调度线程池、文件作业线程池和外部任务队列/图片执行池。
|
||||
* 三个执行池各自独立配置、独立命名、容量互不影响：调度池
|
||||
* （aiimage.scheduling.pool-size，默认 4）与文件作业派发池
|
||||
* （aiimage.result-file-job.*，默认 2 线程/队列 200）互不共享线程；
|
||||
* 外部 Coze 池以虚拟线程 + 信号量限流（默认 12）。容量非法值统一
|
||||
* 外部任务队列池以虚拟线程 + 信号量限流（默认 12）。容量非法值统一
|
||||
* 钳制到最小值；任务失败后信号量名额与调度槽位必须释放，任一池打满
|
||||
* 不影响其他池。
|
||||
*/
|
||||
@@ -55,14 +55,14 @@ class ThreadPoolIsolationConfigTest {
|
||||
return executor;
|
||||
}
|
||||
|
||||
private ExecutorService newCozeVirtual() {
|
||||
ExecutorService executor = new TaskFileJobConfig().cozeVirtualThreadExecutor();
|
||||
private ExecutorService newTaskQueueVirtual() {
|
||||
ExecutorService executor = new TaskFileJobConfig().taskQueueVirtualThreadExecutor();
|
||||
closeables.add(() -> executor.shutdownNow());
|
||||
return executor;
|
||||
}
|
||||
|
||||
private TaskExecutor newCoze(ExecutorService virtualExecutor, int maxConcurrent) {
|
||||
return new TaskFileJobConfig().cozeTaskExecutor(virtualExecutor, maxConcurrent, 4, null);
|
||||
private TaskExecutor newTaskQueue(ExecutorService virtualExecutor, int maxConcurrent) {
|
||||
return new TaskFileJobConfig().taskQueueExecutor(virtualExecutor, maxConcurrent, 4, null);
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -78,15 +78,15 @@ class ThreadPoolIsolationConfigTest {
|
||||
assertEquals(2, dispatch.getMaxPoolSize(), "文件作业池 core=max，不随压力扩张");
|
||||
assertEquals(200, dispatch.getQueueCapacity());
|
||||
|
||||
ExecutorService cozeVirtual = newCozeVirtual();
|
||||
TaskExecutor coze = newCoze(cozeVirtual, 12);
|
||||
ExecutorService taskQueueVirtual = newTaskQueueVirtual();
|
||||
TaskExecutor taskQueue = newTaskQueue(taskQueueVirtual, 12);
|
||||
AtomicBoolean ran = new AtomicBoolean(false);
|
||||
CountDownLatch done = new CountDownLatch(1);
|
||||
coze.execute(() -> {
|
||||
taskQueue.execute(() -> {
|
||||
ran.set(true);
|
||||
done.countDown();
|
||||
});
|
||||
assertTrue(done.await(5, TimeUnit.SECONDS), "Coze 池默认限流 12，正常提交即执行");
|
||||
assertTrue(done.await(5, TimeUnit.SECONDS), "任务队列池默认限流 12，正常提交即执行");
|
||||
assertTrue(ran.get());
|
||||
}
|
||||
|
||||
@@ -96,8 +96,8 @@ class ThreadPoolIsolationConfigTest {
|
||||
// 线程名前缀互不相同，线程转储可识别归属池。
|
||||
ThreadPoolTaskScheduler scheduler = newScheduler(6);
|
||||
ThreadPoolTaskExecutor dispatch = newDispatch(3, 500);
|
||||
ExecutorService cozeVirtual = newCozeVirtual();
|
||||
TaskExecutor coze = newCoze(cozeVirtual, 8);
|
||||
ExecutorService taskQueueVirtual = newTaskQueueVirtual();
|
||||
TaskExecutor taskQueue = newTaskQueue(taskQueueVirtual, 8);
|
||||
assertEquals(6, scheduler.getScheduledThreadPoolExecutor().getCorePoolSize());
|
||||
assertEquals(3, dispatch.getCorePoolSize());
|
||||
assertEquals(500, dispatch.getQueueCapacity());
|
||||
@@ -105,7 +105,7 @@ class ThreadPoolIsolationConfigTest {
|
||||
CountDownLatch all = new CountDownLatch(3);
|
||||
dispatch.execute(all::countDown);
|
||||
scheduler.schedule((Runnable) all::countDown, new Date(System.currentTimeMillis() + 50));
|
||||
coze.execute(all::countDown);
|
||||
taskQueue.execute(all::countDown);
|
||||
assertTrue(all.await(5, TimeUnit.SECONDS), "三个池同时执行互不阻塞");
|
||||
|
||||
assertNotEquals(scheduler.getThreadNamePrefix(), dispatch.getThreadNamePrefix(),
|
||||
@@ -139,11 +139,11 @@ class ThreadPoolIsolationConfigTest {
|
||||
ThreadPoolTaskExecutor dispatch = newDispatch(0, 0);
|
||||
assertEquals(1, dispatch.getCorePoolSize(), "文件作业池 0 钳制到 1");
|
||||
assertEquals(10, dispatch.getQueueCapacity(), "队列 0 钳制到 10");
|
||||
ExecutorService cozeVirtual = newCozeVirtual();
|
||||
TaskExecutor coze = newCoze(cozeVirtual, 0);
|
||||
ExecutorService taskQueueVirtual = newTaskQueueVirtual();
|
||||
TaskExecutor taskQueue = newTaskQueue(taskQueueVirtual, 0);
|
||||
CountDownLatch done = new CountDownLatch(1);
|
||||
coze.execute(done::countDown);
|
||||
assertTrue(done.await(5, TimeUnit.SECONDS), "Coze 池 0 钳制到 1 后仍可执行");
|
||||
taskQueue.execute(done::countDown);
|
||||
assertTrue(done.await(5, TimeUnit.SECONDS), "任务队列池 0 钳制到 1 后仍可执行");
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -200,23 +200,23 @@ class ThreadPoolIsolationConfigTest {
|
||||
assertEquals(1, dispatch.getCorePoolSize());
|
||||
assertEquals(10, dispatch.getQueueCapacity());
|
||||
|
||||
ExecutorService cozeVirtual = newCozeVirtual();
|
||||
TaskExecutor coze = newCoze(cozeVirtual, -3);
|
||||
assertThrows(IllegalArgumentException.class, () -> coze.execute(null), "null 任务被拒绝");
|
||||
ExecutorService taskQueueVirtual = newTaskQueueVirtual();
|
||||
TaskExecutor taskQueue = newTaskQueue(taskQueueVirtual, -3);
|
||||
assertThrows(IllegalArgumentException.class, () -> taskQueue.execute(null), "null 任务被拒绝");
|
||||
CountDownLatch done = new CountDownLatch(1);
|
||||
coze.execute(done::countDown);
|
||||
taskQueue.execute(done::countDown);
|
||||
assertTrue(done.await(5, TimeUnit.SECONDS), "非法配置钳制后池仍可用");
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_074_image_dispatch_job_dependency_failure_releases_resources() throws Exception {
|
||||
// 依赖失败：Coze 任务抛异常后信号量名额必须释放（后续任务可执行）；
|
||||
// 依赖失败：任务队列任务抛异常后信号量名额必须释放（后续任务可执行）；
|
||||
// 调度任务异常被 error handler 吞掉，调度器继续可用。
|
||||
ExecutorService cozeVirtual = newCozeVirtual();
|
||||
TaskExecutor coze = newCoze(cozeVirtual, 2);
|
||||
ExecutorService taskQueueVirtual = newTaskQueueVirtual();
|
||||
TaskExecutor taskQueue = newTaskQueue(taskQueueVirtual, 2);
|
||||
CountDownLatch blockerHeld = new CountDownLatch(1);
|
||||
CountDownLatch releaseBlocker = new CountDownLatch(1);
|
||||
coze.execute(() -> {
|
||||
taskQueue.execute(() -> {
|
||||
blockerHeld.countDown();
|
||||
try {
|
||||
releaseBlocker.await(10, TimeUnit.SECONDS);
|
||||
@@ -225,11 +225,11 @@ class ThreadPoolIsolationConfigTest {
|
||||
}
|
||||
});
|
||||
assertTrue(blockerHeld.await(5, TimeUnit.SECONDS), "任务 1 占住一个信号量名额");
|
||||
coze.execute(() -> {
|
||||
throw new IllegalStateException("coze down");
|
||||
taskQueue.execute(() -> {
|
||||
throw new IllegalStateException("task queue down");
|
||||
});
|
||||
CountDownLatch afterFailure = new CountDownLatch(1);
|
||||
coze.execute(afterFailure::countDown);
|
||||
taskQueue.execute(afterFailure::countDown);
|
||||
assertTrue(afterFailure.await(5, TimeUnit.SECONDS), "失败任务释放名额，后续任务可执行");
|
||||
releaseBlocker.countDown();
|
||||
|
||||
|
||||
Reference in New Issue
Block a user