fix(es): paperuselog state terms 过滤结构修正——{"terms":{"state":[2,4]}} 误写成嵌套 terms 对象被 ES 7.7.1 以 400 拒绝（生产实测）；补 Gateway impl 层单测锁请求体结构

2026-09-22 14:38:44 +08:00
parent c7b6f84db4
commit a61040e2bc
2 changed files with 140 additions and 1 deletions
@@ -45,7 +45,8 @@ public class PaperUseLogEsGatewayImpl implements PaperUseLogEsGateway {
                                  Long fromMillis, Long toMillis, Integer deviceType, String positionMode) {
        ArrayNode filters = baseFilters(districtId, gender, fromMillis, toMillis, deviceType, positionMode);
        // 上纸(2)+撤纸(4) 合并后 rfid 去重（老 :302-342）
-        ArrayNode stateTerms = filters.addObject().putObject("terms").putObject("state").putArray("terms");
+        // 正确结构是 {"terms":{"state":[2,4]}}；写成 {"terms":{"state":{"terms":[...]}}} 会被 ES 以 400 拒绝（2026-09-22 生产实测）
+        ArrayNode stateTerms = filters.addObject().putObject("terms").putArray("state");
        stateTerms.add(2).add(4);
        try {
            ObjectNode body = mapper.createObjectNode();
@@ -0,0 +1,138 @@
+package me.iiv.agent.es;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.io.IOException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class PaperUseLogEsGatewayImplTest {
+
+    @Mock
+    RestClient cloudRestClient;
+
+    @Mock
+    RestClient localRestClient;
+
+    PaperUseLogEsGatewayImpl gateway;
+
+    private final ObjectMapper mapper = new ObjectMapper();
+
+    @BeforeEach
+    void setUp() {
+        gateway = new PaperUseLogEsGatewayImpl(cloudRestClient, localRestClient);
+    }
+
+    private static Response esResponse(String jsonBody) throws IOException {
+        Response response = org.mockito.Mockito.mock(Response.class);
+        when(response.getEntity()).thenReturn(new StringEntity(jsonBody, ContentType.APPLICATION_JSON));
+        return response;
+    }
+
+    private JsonNode captureRequestBody(String responseJson, boolean countOnly) throws Exception {
+        Response response = esResponse(responseJson);
+        when(localRestClient.performRequest(any(Request.class))).thenReturn(response);
+        if (countOnly) {
+            gateway.countState2("paperuselog", true, 2578L, 0,
+                    1789392000000L, 1789478399000L, null, null);
+        } else {
+            gateway.distinctRfidCount("paperuselog", true, 2578L, 0,
+                    1789392000000L, 1789478399000L, null, null);
+        }
+        ArgumentCaptor<Request> captor = ArgumentCaptor.forClass(Request.class);
+        verify(localRestClient).performRequest(captor.capture());
+        Request captured = captor.getValue();
+        assertEquals("POST", captured.getMethod());
+        assertEquals(countOnly ? "/paperuselog/_count" : "/paperuselog/_search", captured.getEndpoint());
+        return mapper.readTree(EntityUtils.toString(captured.getEntity()));
+    }
+
+    @Test
+    void distinctRfidBuildsFlatStateTermsArray() throws Exception {
+        String response = "{\"aggregations\":{\"rfids\":{\"buckets\":[{\"key\":\"a\"},{\"key\":\"b\"}]}}}";
+        JsonNode body = captureRequestBody(response, false);
+
+        assertEquals(0, body.get("size").asInt());
+        assertTrue(!body.get("track_total_hits").asBoolean());
+        // 回归：必须是扁平数组 {"terms":{"state":[2,4]}}；
+        // 嵌套 {"terms":{"state":{"terms":[...]}}} 会被 ES 7.7.1 以 400 "terms_lookup unknown field [terms]" 拒绝（2026-09-22 生产实测）
+        JsonNode stateTerms = body.at("/query/bool/filter").findValue("state");
+        assertTrue(stateTerms != null && stateTerms.isArray());
+        assertEquals(2, stateTerms.size());
+        assertEquals(2, stateTerms.get(0).asInt());
+        assertEquals(4, stateTerms.get(1).asInt());
+        // rfid 是 text 类型，terms 聚合必须走 rfid.keyword（直接聚合 text 报 400）
+        assertEquals("rfid.keyword", body.at("/aggs/rfids/terms/field").asText());
+        assertEquals(10000, body.at("/aggs/rfids/terms/size").asInt());
+        // 基础过滤：districtId/gender term + createTime 闭区间 range（epoch 毫秒）
+        assertTrue(body.at("/query/bool/filter").toString().contains("\"districtId\":{\"value\":2578}"));
+        assertTrue(body.at("/query/bool/filter").toString().contains("\"gender\":{\"value\":0}"));
+        assertEquals(1789392000000L, body.at("/query/bool/filter").findValue("gte").asLong());
+        assertEquals(1789478399000L, body.at("/query/bool/filter").findValue("lte").asLong());
+    }
+
+    @Test
+    void distinctRfidMapsBucketCount() throws Exception {
+        String response = "{\"aggregations\":{\"rfids\":{\"buckets\":[{\"key\":\"a\"},{\"key\":\"b\"},{\"key\":\"c\"}]}}}";
+        Response esResponse = esResponse(response);
+        when(localRestClient.performRequest(any(Request.class))).thenReturn(esResponse);
+
+        long count = gateway.distinctRfidCount("paperuselog", true, 2578L, 0,
+                1789392000000L, 1789478399000L, null, null);
+
+        assertEquals(3L, count);
+    }
+
+    @Test
+    void countState2BuildsTermFilterAndMapsCount() throws Exception {
+        JsonNode body = captureRequestBody("{\"count\":42}", true);
+
+        assertTrue(body.at("/query/bool/filter").toString().contains("\"state\":{\"value\":2}"));
+        Response countResponse = esResponse("{\"count\":42}");
+        when(localRestClient.performRequest(any(Request.class))).thenReturn(countResponse);
+        long count = gateway.countState2("paperuselog", true, 2578L, 0,
+                1789392000000L, 1789478399000L, null, null);
+        assertEquals(42L, count);
+    }
+
+    @Test
+    void routesToCloudClientWhenNotLocalCluster() throws Exception {
+        Response esResponse = esResponse("{\"aggregations\":{\"rfids\":{\"buckets\":[]}}}");
+        when(cloudRestClient.performRequest(any(Request.class))).thenReturn(esResponse);
+
+        gateway.distinctRfidCount("paperuselog", false, 2578L, 0,
+                1789392000000L, 1789478399000L, null, null);
+
+        verify(cloudRestClient).performRequest(any(Request.class));
+    }
+
+    @Test
+    void wrapsEsFailureInIllegalState() throws Exception {
+        when(localRestClient.performRequest(any(Request.class)))
+                .thenThrow(new IOException("connection reset"));
+
+        IllegalStateException exception = assertThrows(IllegalStateException.class,
+                () -> gateway.distinctRfidCount("paperuselog", true, 2578L, 0,
+                        1789392000000L, 1789478399000L, null, null));
+        assertTrue(exception.getMessage().startsWith("ES 查询换纸 rfid 去重失败"));
+    }
+}