fix(mq-starter-redis-stream): 毒消息日志 ERROR 只留摘要,完整堆栈降到 DEBUG
反序列化失败的毒消息按条 ACK 丢弃时,原来整段堆栈打进 ERROR,刷屏淹没有效告警; 改为 ERROR 一行摘要(key + 异常类型 + message),堆栈保留在 DEBUG 供深挖,附护栏单测。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -101,7 +101,10 @@ public class RedisStreamMqDriver implements MqDriver {
|
|||||||
String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), fields.get(StreamMessageConstant.F_BODY));
|
String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), fields.get(StreamMessageConstant.F_BODY));
|
||||||
payload = JSON.parseObject(json, sub.getPayloadType());
|
payload = JSON.parseObject(json, sub.getPayloadType());
|
||||||
} catch (Exception poison) {
|
} 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,不卡队列
|
return true; // 毒消息直接 ACK,不卡队列
|
||||||
}
|
}
|
||||||
MqMessage msg = MqMessage.of(fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), payload);
|
MqMessage msg = MqMessage.of(fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), payload);
|
||||||
|
|||||||
@@ -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<ILoggingEvent> 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<String, String> 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<ILoggingEvent> 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<ILoggingEvent> 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<ILoggingEvent> events(Level level) {
|
||||||
|
return logAppender.list.stream()
|
||||||
|
.filter(e -> e.getLevel() == level)
|
||||||
|
.collect(Collectors.toList());
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user