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()); + } +}