From aa5a949cb9cf4befae2726d3270ad6a460460520 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Thu, 16 Jul 2026 10:24:01 +0100 Subject: [PATCH] =?UTF-8?q?fix(mq-starter-redis-stream):=20=E6=AF=92?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E6=97=A5=E5=BF=97=20ERROR=20=E5=8F=AA?= =?UTF-8?q?=E7=95=99=E6=91=98=E8=A6=81=EF=BC=8C=E5=AE=8C=E6=95=B4=E5=A0=86?= =?UTF-8?q?=E6=A0=88=E9=99=8D=E5=88=B0=20DEBUG?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 反序列化失败的毒消息按条 ACK 丢弃时，原来整段堆栈打进 ERROR，刷屏淹没有效告警； 改为 ERROR 一行摘要（key + 异常类型 + message），堆栈保留在 DEBUG 供深挖，附护栏单测。 Co-Authored-By: Claude Fable 5 --- .../mq/driver/redis/RedisStreamMqDriver.java | 5 +- .../RedisStreamMqDriverPoisonLogTest.java | 82 +++++++++++++++++++ 2 files changed, 86 insertions(+), 1 deletion(-) create mode 100644 mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverPoisonLogTest.java diff --git a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java index ab918f6..fa5aa7e 100644 --- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java +++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java @@ -101,7 +101,10 @@ public class RedisStreamMqDriver implements MqDriver { String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), fields.get(StreamMessageConstant.F_BODY)); payload = JSON.parseObject(json, sub.getPayloadType()); } catch (Exception poison) { - log.error("[mq-redis] 消息反序列化失败，按毒消息 ACK 丢弃 key={}", fields.get(StreamMessageConstant.F_KEY), poison); + String key = fields.get(StreamMessageConstant.F_KEY); + log.error("[mq-redis] 消息反序列化失败，按毒消息 ACK 丢弃 key={} type={} message={}", + key, poison.getClass().getSimpleName(), poison.getMessage()); + log.debug("[mq-redis] 消息反序列化失败堆栈 key={}", key, poison); return true; // 毒消息直接 ACK，不卡队列 } MqMessage msg = MqMessage.of(fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), payload); diff --git a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverPoisonLogTest.java b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverPoisonLogTest.java new file mode 100644 index 0000000..c4462f7 --- /dev/null +++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverPoisonLogTest.java @@ -0,0 +1,82 @@ +package com.njcn.mq.driver.redis; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; +import com.njcn.middle.stream.constant.StreamMessageConstant; +import com.njcn.mq.spi.MqSubscription; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.Method; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.*; + +class RedisStreamMqDriverPoisonLogTest { + + private final ListAppender logAppender = new ListAppender<>(); + private Logger logger; + private Level originalLevel; + + @BeforeEach + void attachAppender() { + logger = (Logger) LoggerFactory.getLogger(RedisStreamMqDriver.class); + originalLevel = logger.getLevel(); + logger.setLevel(Level.DEBUG); + logAppender.start(); + logger.addAppender(logAppender); + } + + @AfterEach + void detachAppender() { + logger.detachAppender(logAppender); + logger.setLevel(originalLevel); + } + + @Test + void poisonMessage_logsErrorSummaryAndDebugStackTrace() throws Exception { + RedisStreamMqDriver driver = new RedisStreamMqDriver(null, null, null); + MqSubscription sub = MqSubscription.builder() + .tag("*") + .payloadType(String.class) + .listener(msg -> fail("poison message must not reach listener")) + .build(); + + Map fields = new HashMap<>(); + fields.put(StreamMessageConstant.F_KEY, "poison-1"); + fields.put(StreamMessageConstant.F_TAG, ""); + fields.put(StreamMessageConstant.F_ENC, "plain"); + fields.put(StreamMessageConstant.F_BODY, "new"); + + Method dispatch = RedisStreamMqDriver.class.getDeclaredMethod("dispatch", MqSubscription.class, Map.class, int.class); + dispatch.setAccessible(true); + + Object ack = dispatch.invoke(driver, sub, fields, 3); + + assertEquals(Boolean.TRUE, ack, "poison message should still be ACKed"); + List errors = events(Level.ERROR); + assertEquals(1, errors.size()); + ILoggingEvent error = errors.get(0); + assertNull(error.getThrowableProxy(), "ERROR summary should not print a stack trace by default"); + assertTrue(error.getFormattedMessage().contains("key=poison-1")); + assertTrue(error.getFormattedMessage().contains("type=JSONException")); + assertTrue(error.getFormattedMessage().contains("message=")); + + List debugs = events(Level.DEBUG); + assertEquals(1, debugs.size()); + assertNotNull(debugs.get(0).getThrowableProxy(), "DEBUG log should retain the stack trace for deep diagnosis"); + } + + private List events(Level level) { + return logAppender.list.stream() + .filter(e -> e.getLevel() == level) + .collect(Collectors.toList()); + } +}