fix(es): ES 集群 7.7.1 与高阶客户端 7.17 不兼容，电量消息/换纸日志查询降级低阶 RestClient

生产两个 ES 集群均为 7.7.1：elasticsearch-java 7.17 的 vendor Content-Type (application/vnd.elasticsearch+json; compatible-with=7) 被 406 拒绝，且其 X-Elastic-Product 响应头校验（7.14 引入）在 7.7.1 上无法通过。 ElectricityMessageEsGatewayImpl 与 PaperUseLogEsGatewayImpl 改用低阶 RestClient 直连（application/json、无产品头校验），查询语义不变。修复REQ-20260921-0007上线后 tools/call 406 冒烟失败。
2026-09-22 11:21:46 +08:00
parent 10d618dac2
commit 6be5169217
4 changed files with 184 additions and 156 deletions
@@ -22,6 +22,8 @@ import java.util.Arrays;
 * 双集群 ES 客户端：cloud=云集群、local=本地化集群（老系统 Dictionary.LocalEsDate 分界）。
 * 低阶 RestClient 也注册为 bean（destroyMethod=close 随容器关闭）；
 * 高阶 ElasticsearchClient 包装同一 RestClient，关闭 client 即关闭底层连接。
+ * 注意：生产两个集群均为 ES 7.7.1，高阶客户端（7.17）的 vendor Content-Type 与
+ * X-Elastic-Product 校验与 7.7.1 不兼容（406），业务查询一律注入低阶 RestClient bean。
 */
@Configuration
 public class EsClientConfig {
@@ -1,16 +1,12 @@
 package me.iiv.agent.es;

-import co.elastic.clients.elasticsearch.ElasticsearchClient;
-import co.elastic.clients.elasticsearch._types.FieldValue;
-import co.elastic.clients.elasticsearch._types.SortOrder;
-import co.elastic.clients.elasticsearch._types.query_dsl.Query;
-import co.elastic.clients.elasticsearch.core.SearchResponse;
-import co.elastic.clients.elasticsearch.core.search.Hit;
-import co.elastic.clients.json.JsonData;
-import jakarta.json.JsonNumber;
-import jakarta.json.JsonObject;
-import jakarta.json.JsonString;
-import jakarta.json.JsonValue;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;

@@ -22,8 +18,12 @@ import java.util.ArrayList;
 import java.util.List;

 /**
- * electricitymessage_yyyyMM ES 查询实现（elasticsearch-java 7.17 强类型 API），
- * 只注入 esLocal 客户端（索引只在本地集群，不走 esCloud）。
+ * electricitymessage_yyyyMM ES 查询实现。
+ * 降级原因：生产 ES 集群为 7.7.1，elasticsearch-java 7.17 高阶客户端（ElasticsearchClient）
+ * 请求自带 vendor Content-Type（application/vnd.elasticsearch+json; compatible-with=7）被
+ * 7.7.1 以 406 Not Acceptable 拒绝，且 7.17 客户端校验响应 X-Elastic-Product 头（7.14 才引入），
+ * 7.7.1 没有，二次兼容也过不了；故改用低阶 RestClient（只发普通 application/json、无产品头校验）。
+ * 只注入 esLocalRestClient（索引只在本地集群，不走 esCloud）。
 * 查询口径对齐老 BatteryAnalyseController.batteryAnalyseList 但修掉其坑：
 * snCode 用 match_phrase（老代码 matchQuery 经 standard analyzer 会把 AB-123 切成 ab/123 误配）、
 * messageType 不传不加 term（老代码默认值 0 会把"不传"歧义成"电量"）、createTime 闭区间 range、
@@ -37,47 +37,50 @@ public class ElectricityMessageEsGatewayImpl implements ElectricityMessageEsGate
    private static final DateTimeFormatter TIME_FORMATTER =
            DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(BUSINESS_ZONE);

-    private final ElasticsearchClient localClient;
+    private final RestClient localRestClient;
+    private final ObjectMapper mapper = new ObjectMapper();

-    public ElectricityMessageEsGatewayImpl(@Qualifier("esLocalClient") ElasticsearchClient localClient) {
-        this.localClient = localClient;
+    public ElectricityMessageEsGatewayImpl(@Qualifier("esLocalRestClient") RestClient localRestClient) {
+        this.localRestClient = localRestClient;
    }

    @Override
    public Result search(String index, String snCode, Integer messageType,
                         String startTime, String endTime, int from, int size) {
-        List<Query> filters = new ArrayList<>();
+        ArrayNode filters = mapper.createArrayNode();
        // 老代码 snCode 字符串走 matchQuery（BaseSearchSupportImpl2 type=0 非数值分支），
        // standard analyzer 会把 AB-123 切成 ab/123 造成误配；改用 match_phrase 精确短语
-        filters.add(Query.of(q -> q.matchPhrase(m -> m.field("snCode").query(snCode))));
+        filters.addObject().putObject("match_phrase").putObject("snCode").put("query", snCode);
        if (messageType != null) {
-            filters.add(Query.of(q -> q.term(t -> t.field("messageType")
-                    .value(FieldValue.of(messageType.longValue())))));
+            filters.addObject().putObject("term").putObject("messageType").put("value", messageType);
        }
        // 老代码 rangeQuery from(start,true).to(end,true)，闭区间
-        filters.add(Query.of(q -> q.range(r -> r.field("createTime")
-                .gte(JsonData.of(startTime)).lte(JsonData.of(endTime)))));
+        ObjectNode range = filters.addObject().putObject("range").putObject("createTime");
+        range.put("gte", startTime).put("lte", endTime);
+
+        ObjectNode body = mapper.createObjectNode();
+        body.put("from", from).put("size", size);
+        body.put("track_total_hits", true);
+        ObjectNode bool = body.putObject("query").putObject("bool");
+        bool.set("filter", filters);
+        body.putArray("sort").addObject().putObject("createTime").put("order", "asc");
        try {
-            SearchResponse<JsonData> response = localClient.search(s -> s
-                            .index(index)
-                            .from(from)
-                            .size(size)
-                            .trackTotalHits(t -> t.enabled(true))
-                            .query(q -> q.bool(b -> b.filter(filters)))
-                            .sort(so -> so.field(f -> f.field("createTime").order(SortOrder.Asc))),
-                    JsonData.class);
+            Request req = new Request("POST", "/" + index + "/_search");
+            // setJsonEntity 自动带 Content-Type: application/json（无 vendor 头）
+            req.setJsonEntity(mapper.writeValueAsString(body));
+            Response resp = localRestClient.performRequest(req);
+            JsonNode root = mapper.readTree(resp.getEntity().getContent());
            List<Record> records = new ArrayList<>();
-            for (Hit<JsonData> hit : response.hits().hits()) {
-                JsonData source = hit.source();
+            for (JsonNode hit : root.at("/hits/hits")) {
+                JsonNode source = hit.get("_source");
                if (source == null) {
                    continue;
                }
-                JsonObject obj = source.toJson().asJsonObject();
-                records.add(new Record(createTimeText(obj.get("createTime")),
-                        integerValue(obj.get("messageType")),
-                        integerValue(obj.get("value"))));
+                records.add(new Record(createTimeText(source.get("createTime")),
+                        integerValue(source.get("messageType")),
+                        integerValue(source.get("value"))));
            }
-            return new Result(response.hits().total().value(), records);
+            return new Result(root.at("/hits/total/value").asLong(), records);
        } catch (IOException | RuntimeException e) {
            throw new IllegalStateException("ES 查询电量消息记录失败: " + e.getMessage(), e);
        }
@@ -87,21 +90,20 @@ public class ElectricityMessageEsGatewayImpl implements ElectricityMessageEsGate
     * createTime 在索引里可能是 yyyy-MM-dd HH:mm:ss 文本（老 range 直接传字符串可证 ES 接受该格式），
     * 也可能是 epoch 毫秒数；两种形态都归一为 yyyy-MM-dd HH:mm:ss 文本输出。
     */
-    private static String createTimeText(JsonValue value) {
-        if (value == null || value.getValueType() == JsonValue.ValueType.NULL) {
+    private static String createTimeText(JsonNode value) {
+        if (value == null || value.isNull()) {
            return null;
        }
-        if (value instanceof JsonString) {
-            return ((JsonString) value).getString();
+        if (value.isTextual()) {
+            return value.asText();
        }
-        if (value instanceof JsonNumber) {
-            return TIME_FORMATTER.format(Instant.ofEpochMilli(((JsonNumber) value).longValue()));
+        if (value.isNumber()) {
+            return TIME_FORMATTER.format(Instant.ofEpochMilli(value.longValue()));
        }
        return null;
    }

-    private static Integer integerValue(JsonValue value) {
-        return value != null && value.getValueType() == JsonValue.ValueType.NUMBER
-                ? ((JsonNumber) value).intValue() : null;
+    private static Integer integerValue(JsonNode value) {
+        return value != null && value.isNumber() ? value.intValue() : null;
    }
 }
@@ -1,19 +1,24 @@
 package me.iiv.agent.es;

-import co.elastic.clients.elasticsearch.ElasticsearchClient;
-import co.elastic.clients.elasticsearch._types.FieldValue;
-import co.elastic.clients.elasticsearch._types.query_dsl.Query;
-import co.elastic.clients.json.JsonData;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 import org.springframework.util.StringUtils;

 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;

 /**
- * paperuselog ES 查询实现（elasticsearch-java 7.17 强类型 API）。
+ * paperuselog ES 查询实现。
+ * 降级原因：生产 ES 集群为 7.7.1，elasticsearch-java 7.17 高阶客户端（ElasticsearchClient）
+ * 请求自带 vendor Content-Type（application/vnd.elasticsearch+json; compatible-with=7）被
+ * 7.7.1 以 406 Not Acceptable 拒绝，且 7.17 客户端校验响应 X-Elastic-Product 头（7.14 才引入），
+ * 7.7.1 没有，二次兼容也过不了；故改用低阶 RestClient（只发普通 application/json、无产品头校验）。
 * 查询口径对齐老 MonthlyStatisticsController.getMonthlyStatisticsMap：
 * term districtId/gender、range createTime [gte,lte]（含边界）、deviceType 用 match（老代码字符串参数走 match），
 * positionMode 厕位=positionsId>0、厕墙=positionsId=0。
@@ -25,31 +30,37 @@ public class PaperUseLogEsGatewayImpl implements PaperUseLogEsGateway {
    /** rfid terms 聚合去重上限，超出部分不计入 */
    private static final int RFID_TERMS_SIZE = 10000;

-    private final ElasticsearchClient cloudClient;
-    private final ElasticsearchClient localClient;
+    private final RestClient cloudRestClient;
+    private final RestClient localRestClient;
+    private final ObjectMapper mapper = new ObjectMapper();

-    public PaperUseLogEsGatewayImpl(@Qualifier("esCloudClient") ElasticsearchClient cloudClient,
-                                    @Qualifier("esLocalClient") ElasticsearchClient localClient) {
-        this.cloudClient = cloudClient;
-        this.localClient = localClient;
+    public PaperUseLogEsGatewayImpl(@Qualifier("esCloudRestClient") RestClient cloudRestClient,
+                                    @Qualifier("esLocalRestClient") RestClient localRestClient) {
+        this.cloudRestClient = cloudRestClient;
+        this.localRestClient = localRestClient;
    }

    @Override
    public long distinctRfidCount(String index, boolean localCluster, Long districtId, Integer gender,
                                  Long fromMillis, Long toMillis, Integer deviceType, String positionMode) {
-        List<Query> filters = baseFilters(districtId, gender, fromMillis, toMillis, deviceType, positionMode);
+        ArrayNode filters = baseFilters(districtId, gender, fromMillis, toMillis, deviceType, positionMode);
        // 上纸(2)+撤纸(4) 合并后 rfid 去重（老 :302-342）
-        filters.add(Query.of(q -> q.terms(t -> t.field("state")
-                .terms(tt -> tt.value(List.of(FieldValue.of(2), FieldValue.of(4)))))));
+        ArrayNode stateTerms = filters.addObject().putObject("terms").putObject("state").putArray("terms");
+        stateTerms.add(2).add(4);
        try {
-            return client(localCluster).search(s -> s
-                            .index(index)
-                            .size(0)
-                            .trackTotalHits(t -> t.enabled(true))
-                            .query(q -> q.bool(b -> b.filter(filters)))
-                            .aggregations("rfids", a -> a.terms(t -> t.field("rfid").size(RFID_TERMS_SIZE))),
-                    Void.class)
-                    .aggregations().get("rfids").sterms().buckets().array().size();
+            ObjectNode body = mapper.createObjectNode();
+            body.put("size", 0);
+            body.put("track_total_hits", false);
+            ObjectNode bool = body.putObject("query").putObject("bool");
+            bool.set("filter", filters);
+            body.putObject("aggs").putObject("rfids").putObject("terms")
+                    .put("field", "rfid").put("size", RFID_TERMS_SIZE);
+
+            Request req = new Request("POST", "/" + index + "/_search");
+            req.setJsonEntity(mapper.writeValueAsString(body));
+            Response resp = client(localCluster).performRequest(req);
+            JsonNode root = mapper.readTree(resp.getEntity().getContent());
+            return root.at("/aggregations/rfids/buckets").size();
        } catch (IOException | RuntimeException e) {
            throw new IllegalStateException("ES 查询换纸 rfid 去重失败: " + e.getMessage(), e);
        }
@@ -58,40 +69,47 @@ public class PaperUseLogEsGatewayImpl implements PaperUseLogEsGateway {
    @Override
    public long countState2(String index, boolean localCluster, Long districtId, Integer gender,
                            Long fromMillis, Long toMillis, Integer deviceType, String positionMode) {
-        List<Query> filters = baseFilters(districtId, gender, fromMillis, toMillis, deviceType, positionMode);
-        filters.add(Query.of(q -> q.term(t -> t.field("state").value(FieldValue.of(2)))));
+        ArrayNode filters = baseFilters(districtId, gender, fromMillis, toMillis, deviceType, positionMode);
+        filters.addObject().putObject("term").putObject("state").put("value", 2);
        try {
-            return client(localCluster).count(c -> c.index(index)
-                    .query(q -> q.bool(b -> b.filter(filters)))).count();
+            ObjectNode body = mapper.createObjectNode();
+            ObjectNode bool = body.putObject("query").putObject("bool");
+            bool.set("filter", filters);
+
+            Request req = new Request("POST", "/" + index + "/_count");
+            req.setJsonEntity(mapper.writeValueAsString(body));
+            Response resp = client(localCluster).performRequest(req);
+            JsonNode root = mapper.readTree(resp.getEntity().getContent());
+            return root.get("count").asLong();
        } catch (IOException | RuntimeException e) {
            throw new IllegalStateException("ES 查询换纸上纸记录数失败: " + e.getMessage(), e);
        }
    }

-    private List<Query> baseFilters(Long districtId, Integer gender, Long fromMillis, Long toMillis,
-                                    Integer deviceType, String positionMode) {
-        List<Query> filters = new ArrayList<>();
-        filters.add(Query.of(q -> q.term(t -> t.field("districtId").value(FieldValue.of(districtId)))));
-        filters.add(Query.of(q -> q.term(t -> t.field("gender").value(FieldValue.of(gender)))));
+    private ArrayNode baseFilters(Long districtId, Integer gender, Long fromMillis, Long toMillis,
+                                  Integer deviceType, String positionMode) {
+        ArrayNode filters = mapper.createArrayNode();
+        filters.addObject().putObject("term").putObject("districtId").put("value", districtId);
+        filters.addObject().putObject("term").putObject("gender").put("value", gender);
        // createTime 为 epoch 毫秒；含边界（老 rangeQuery from/to 均 inclusive）
-        filters.add(Query.of(q -> q.range(r -> r.field("createTime")
-                .gte(JsonData.of(fromMillis)).lte(JsonData.of(toMillis)))));
+        ObjectNode range = filters.addObject().putObject("range").putObject("createTime");
+        range.put("gte", fromMillis).put("lte", toMillis);
        if (deviceType != null) {
            // 老代码 deviceType 为字符串参数走 matchQuery（BaseSearchSupportImpl2 type=0 非数值分支）
-            filters.add(Query.of(q -> q.match(m -> m.field("deviceType")
-                    .query(FieldValue.of(String.valueOf(deviceType))))));
+            filters.addObject().putObject("match").putObject("deviceType")
+                    .put("query", String.valueOf(deviceType));
        }
        if (StringUtils.hasText(positionMode)) {
            if ("position".equals(positionMode)) {
-                filters.add(Query.of(q -> q.range(r -> r.field("positionsId").gt(JsonData.of(0)))));
+                filters.addObject().putObject("range").putObject("positionsId").put("gt", 0);
            } else {
-                filters.add(Query.of(q -> q.term(t -> t.field("positionsId").value(FieldValue.of(0)))));
+                filters.addObject().putObject("term").putObject("positionsId").put("value", 0);
            }
        }
        return filters;
    }

-    private ElasticsearchClient client(boolean localCluster) {
-        return localCluster ? localClient : cloudClient;
+    private RestClient client(boolean localCluster) {
+        return localCluster ? localRestClient : cloudRestClient;
    }
 }
@@ -1,12 +1,13 @@
 package me.iiv.agent.es;

-import co.elastic.clients.elasticsearch.ElasticsearchClient;
-import co.elastic.clients.elasticsearch.core.SearchRequest;
-import co.elastic.clients.elasticsearch.core.SearchResponse;
-import co.elastic.clients.elasticsearch.core.search.Hit;
-import co.elastic.clients.elasticsearch.core.search.TotalHitsRelation;
-import co.elastic.clients.json.JsonData;
-import co.elastic.clients.util.ObjectBuilder;
+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;
@@ -14,103 +15,107 @@ import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;

-import java.util.List;
-import java.util.function.Function;
+import java.io.IOException;

 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
 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.ArgumentMatchers.eq;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;

@ExtendWith(MockitoExtension.class)
 class ElectricityMessageEsGatewayImplTest {

+    /** 2026-09-21 14:00:00 Asia/Shanghai 的 epoch 毫秒（= 2026-09-21T06:00:00Z） */
+    private static final long EPOCH_MILLIS = 1789970400000L;
+
+    private static final String RESPONSE_JSON = "{"
+            + "\"took\":1,\"timed_out\":false,"
+            + "\"_shards\":{\"total\":1,\"successful\":1,\"skipped\":0,\"failed\":0},"
+            + "\"hits\":{\"total\":{\"value\":2,\"relation\":\"eq\"},"
+            + "\"hits\":["
+            + "{\"_index\":\"electricitymessage_202609\","
+            + "\"_source\":{\"createTime\":\"2026-09-21 10:00:00\",\"messageType\":0,\"value\":88}},"
+            + "{\"_index\":\"electricitymessage_202609\","
+            + "\"_source\":{\"createTime\":" + EPOCH_MILLIS + ",\"messageType\":4,\"value\":3700}}"
+            + "]}}";
+
    @Mock
-    ElasticsearchClient localClient;
+    RestClient localRestClient;

    ElectricityMessageEsGatewayImpl gateway;

+    private final ObjectMapper mapper = new ObjectMapper();
+
    @BeforeEach
    void setUp() {
-        gateway = new ElectricityMessageEsGatewayImpl(localClient);
+        gateway = new ElectricityMessageEsGatewayImpl(localRestClient);
    }

-    private SearchResponse<JsonData> responseWith(String... sources) {
-        List<Hit<JsonData>> hits = java.util.Arrays.stream(sources)
-                .map(json -> Hit.<JsonData>of(hit -> hit
-                        .index("electricitymessage_202609")
-                        .source(JsonData.fromJson(json))))
-                .toList();
-        return SearchResponse.of(r -> r
-                .took(1)
-                .timedOut(false)
-                .shards(s -> s.total(1).successful(1).skipped(0).failed(0))
-                .hits(h -> h.total(t -> t.value(2).relation(TotalHitsRelation.Eq))
-                        .hits(hits)));
+    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;
    }

-    @SuppressWarnings("unchecked")
-    private SearchRequest captureRequest(SearchResponse<JsonData> response, Integer messageType) throws Exception {
-        when(localClient.search(any(Function.class), eq(JsonData.class))).thenReturn(response);
+    private JsonNode captureRequestBody(String responseJson, Integer messageType) throws Exception {
+        Response esResponse = esResponse(responseJson);
+        when(localRestClient.performRequest(any(Request.class))).thenReturn(esResponse);
        gateway.search("electricitymessage_202609", "AB-123", messageType,
                "2026-09-21 00:00:00", "2026-09-21 23:59:59", 0, 500);
-        ArgumentCaptor<Function<SearchRequest.Builder, ObjectBuilder<SearchRequest>>> captor =
-                ArgumentCaptor.forClass(Function.class);
-        verify(localClient).search(captor.capture(), eq(JsonData.class));
-        return captor.getValue().apply(new SearchRequest.Builder()).build();
+        ArgumentCaptor<Request> captor = ArgumentCaptor.forClass(Request.class);
+        verify(localRestClient).performRequest(captor.capture());
+        Request captured = captor.getValue();
+        assertEquals("POST", captured.getMethod());
+        assertEquals("/electricitymessage_202609/_search", captured.getEndpoint());
+        return mapper.readTree(EntityUtils.toString(captured.getEntity()));
+    }
+
+    private static boolean anyFilter(JsonNode filters, java.util.function.Predicate<JsonNode> predicate) {
+        for (JsonNode filter : filters) {
+            if (predicate.test(filter)) {
+                return true;
+            }
+        }
+        return false;
    }

    @Test
-    @SuppressWarnings("unchecked")
    void buildsMatchPhraseOptionalTermRangeSortAndPagination() throws Exception {
-        SearchRequest request = captureRequest(responseWith(
-                "{\"createTime\":\"2026-09-21 10:00:00\",\"messageType\":0,\"value\":88}"), 0);
+        JsonNode body = captureRequestBody(RESPONSE_JSON, 0);

-        assertEquals(List.of("electricitymessage_202609"), request.index());
-        assertEquals(Integer.valueOf(0), request.from());
-        assertEquals(Integer.valueOf(500), request.size());
-        assertTrue(request.trackTotalHits().enabled());
-        assertEquals(1, request.sort().size());
-        assertEquals("createTime", request.sort().get(0).field().field());
-        assertEquals(co.elastic.clients.elasticsearch._types.SortOrder.Asc,
-                request.sort().get(0).field().order());
+        assertEquals(0, body.get("from").asInt());
+        assertEquals(500, body.get("size").asInt());
+        assertTrue(body.get("track_total_hits").asBoolean());
+        assertEquals("asc", body.get("sort").get(0).get("createTime").get("order").asText());

-        List<co.elastic.clients.elasticsearch._types.query_dsl.Query> filters =
-                request.query().bool().filter();
+        JsonNode filters = body.at("/query/bool/filter");
        // snCode 必须 match_phrase（老系统 matchQuery 分词误配的修复）
-        assertTrue(filters.stream().anyMatch(q -> q.isMatchPhrase()
-                && q.matchPhrase().field().equals("snCode")
-                && q.matchPhrase().query().equals("AB-123")));
+        assertTrue(anyFilter(filters, f -> f.has("match_phrase")
+                && f.get("match_phrase").get("snCode").get("query").asText().equals("AB-123")));
        // messageType 传了才加 term
-        assertTrue(filters.stream().anyMatch(q -> q.isTerm()
-                && q.term().field().equals("messageType")
-                && q.term().value().longValue() == 0));
+        assertTrue(anyFilter(filters, f -> f.has("term")
+                && f.get("term").get("messageType").get("value").asInt() == 0));
        // createTime 闭区间 range
-        assertTrue(filters.stream().anyMatch(q -> q.isRange()
-                && q.range().field().equals("createTime")
-                && q.range().gte().to(String.class).equals("2026-09-21 00:00:00")
-                && q.range().lte().to(String.class).equals("2026-09-21 23:59:59")));
+        assertTrue(anyFilter(filters, f -> f.has("range")
+                && f.get("range").get("createTime").get("gte").asText().equals("2026-09-21 00:00:00")
+                && f.get("range").get("createTime").get("lte").asText().equals("2026-09-21 23:59:59")));
    }

    @Test
-    @SuppressWarnings("unchecked")
    void omitsMessageTypeTermWhenNull() throws Exception {
-        SearchRequest request = captureRequest(responseWith(
-                "{\"createTime\":\"2026-09-21 10:00:00\",\"messageType\":1,\"value\":-60}"), null);
-        assertTrue(request.query().bool().filter().stream().noneMatch(q -> q.isTerm()));
+        JsonNode body = captureRequestBody(RESPONSE_JSON, null);
+        for (JsonNode filter : body.at("/query/bool/filter")) {
+            assertTrue(!filter.has("term") || filter.get("term").findValue("messageType") == null);
+        }
    }

    @Test
-    @SuppressWarnings("unchecked")
    void mapsHitsAndEpochMillisCreateTime() throws Exception {
-        long epochMillis = java.time.Instant.parse("2026-09-21T06:00:00Z").toEpochMilli();
-        when(localClient.search(any(Function.class), eq(JsonData.class))).thenReturn(responseWith(
-                "{\"createTime\":\"2026-09-21 10:00:00\",\"messageType\":0,\"value\":88}",
-                "{\"createTime\":" + epochMillis + ",\"messageType\":4,\"value\":3700}"));
+        Response esResponse = esResponse(RESPONSE_JSON);
+        when(localRestClient.performRequest(any(Request.class))).thenReturn(esResponse);

        ElectricityMessageEsGateway.Result result = gateway.search("electricitymessage_202609",
                "AB-123", null, "2026-09-21 00:00:00", "2026-09-21 23:59:59", 0, 500);
@@ -124,13 +129,15 @@ class ElectricityMessageEsGatewayImplTest {
        // epoch 毫秒形态归一为 Asia/Shanghai yyyy-MM-dd HH:mm:ss
        assertEquals("2026-09-21 14:00:00", result.getRecords().get(1).getCreateTime());
        assertEquals(Integer.valueOf(4), result.getRecords().get(1).getMessageType());
+        assertEquals(Integer.valueOf(3700), result.getRecords().get(1).getValue());
    }

    @Test
-    @SuppressWarnings("unchecked")
    void mapsNullAndMissingFields() throws Exception {
-        when(localClient.search(any(Function.class), eq(JsonData.class))).thenReturn(responseWith(
-                "{\"createTime\":null}"));
+        String response = "{\"hits\":{\"total\":{\"value\":1,\"relation\":\"eq\"},"
+                + "\"hits\":[{\"_source\":{\"createTime\":null}}]}}";
+        Response esResponse = esResponse(response);
+        when(localRestClient.performRequest(any(Request.class))).thenReturn(esResponse);

        ElectricityMessageEsGateway.Record record = gateway.search("electricitymessage_202609",
                "AB-123", null, "2026-09-21 00:00:00", "2026-09-21 23:59:59", 0, 500)
@@ -142,10 +149,9 @@ class ElectricityMessageEsGatewayImplTest {
    }

    @Test
-    @SuppressWarnings("unchecked")
    void wrapsEsFailureInIllegalState() throws Exception {
-        when(localClient.search(any(Function.class), eq(JsonData.class)))
-                .thenThrow(new java.io.IOException("connection reset"));
+        when(localRestClient.performRequest(any(Request.class)))
+                .thenThrow(new IOException("connection reset"));

        IllegalStateException exception = assertThrows(IllegalStateException.class,
                () -> gateway.search("electricitymessage_202609", "AB-123", null,