feat(mq-starter-redis-stream): reclaim 耗尽分支也过滤 tag，不归本组的消息不落库

tag 不匹配的高 delivery 残留（反复崩溃场景）直接 ACK 丢弃，
不回调 retryExhaustedHandler——不得以本组名义当"重试耗尽"落库。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java

@@ -146,6 +146,13 @@ public class RedisStreamMqDriver implements MqDriver {
             Map<String, String> fields = new HashMap<>();
             rec.getValue().forEach((k, v) -> fields.put(
                     new String(k, StandardCharsets.UTF_8), new String(v, StandardCharsets.UTF_8)));
+            if (!MqTagMatcher.matches(sub.getTag(), fields.get(StreamMessageConstant.F_TAG))) {
+                // tag 不匹配：不归本组消费，直接 ACK 丢弃——不得进耗尽分支以本组名义落库，也无须重投。
+                redis.opsForStream().acknowledge(stream, group, rec.getId());
+                log.debug("[mq-redis] tag 不匹配（reclaim），ACK 丢弃 key={} delivery={}",
+                        fields.get(StreamMessageConstant.F_KEY), deliveryCount);
+                continue;
+            }
             if (deliveryCount > maxRetry) {
                 if (sub.getRetryExhaustedHandler() != null) {
                     try {

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverTagFilterIT.java

@@ -6,15 +6,21 @@ import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
 import com.njcn.middle.stream.support.StreamKeyBuilder;
 import org.junit.jupiter.api.Test;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
-import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
+import org.springframework.data.redis.connection.stream.*;
+import org.springframework.data.redis.core.RedisCallback;
 import org.springframework.data.redis.core.StringRedisTemplate;
 
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
 class RedisStreamMqDriverTagFilterIT {
@@ -51,6 +57,59 @@ class RedisStreamMqDriverTagFilterIT {
         driver.stop();
     }
 
+    @Test
+    void reclaimExhausted_mismatchedTag_ackedWithoutExhaustedCallback() throws Exception {
+        StringRedisTemplate redis = newRedisOrSkip();
+        StreamKeyBuilder kb = new StreamKeyBuilder(false, "");
+        RedisStreamProperties props = new RedisStreamProperties();
+        props.setConsumerName("it-tag-reclaim");
+        props.setReadCount(10);
+        props.setBlockMs(300);
+        props.setMaxRetry(1);
+        RedisStreamMqDriver driver = new RedisStreamMqDriver(redis, kb, props, 200L);
+
+        String topic = "MQ_TAG_RECLAIM_IT_" + System.nanoTime();
+        String stream = kb.buildStream(topic);
+
+        // 1. 手工建组（driver 尚未 subscribe，模拟历史遗留 PEL 场景）
+        redis.execute((RedisCallback<Object>) c -> c.execute("XGROUP",
+                "CREATE".getBytes(StandardCharsets.UTF_8), stream.getBytes(StandardCharsets.UTF_8),
+                "g1".getBytes(StandardCharsets.UTF_8), "$".getBytes(StandardCharsets.UTF_8),
+                "MKSTREAM".getBytes(StandardCharsets.UTF_8)));
+
+        // 2. 发一条 tag=B 的消息（send 不依赖 subscribe）
+        driver.send(topic, MqMessage.of("k-stale", "B", "stale"));
+
+        // 3. 手工 XREADGROUP 不 ACK：delivery=1，模拟消费者读走后崩溃
+        List<MapRecord<String, Object, Object>> recs = redis.opsForStream().read(
+                Consumer.from("g1", "seed"),
+                StreamReadOptions.empty().count(1),
+                StreamOffset.create(stream, ReadOffset.lastConsumed()));
+        assertNotNull(recs);
+        assertEquals(1, recs.size(), "种子消息应已进 PEL");
+        RecordId staleId = recs.get(0).getId();
+
+        // 4. 手工 XCLAIM 一次：delivery=2 > maxRetry=1，reclaim 将走耗尽分支
+        redis.execute((RedisCallback<Object>) conn -> conn.streamCommands()
+                .xClaim(stream.getBytes(StandardCharsets.UTF_8), "g1", "seed2", Duration.ZERO, staleId));
+
+        // 5. 启动订阅 tag=A 的 driver，等 reclaim 处理这条 tag=B 的残留
+        BlockingQueue<MqMessage> exhausted = new LinkedBlockingQueue<>();
+        BlockingQueue<Object> delivered = new LinkedBlockingQueue<>();
+        driver.subscribe(MqSubscription.builder()
+                .topic(topic).group("g1").tag("A").payloadType(String.class)
+                .concurrency(1).batchSize(1).maxRetry(1)
+                .listener(m -> delivered.add(m.getPayload()))
+                .retryExhaustedHandler(exhausted::add)
+                .build());
+
+        assertNull(exhausted.poll(3, TimeUnit.SECONDS), "tag 不匹配的消息不得以本组名义耗尽落库");
+        assertTrue(delivered.isEmpty(), "tag 不匹配的消息不得进业务方法");
+        PendingMessagesSummary summary = redis.opsForStream().pending(stream, "g1");
+        assertEquals(0L, summary.getTotalPendingMessages(), "应被 ACK 丢弃，PEL 清空");
+        driver.stop();
+    }
+
     private static StringRedisTemplate newRedisOrSkip() {
         try {
             LettuceConnectionFactory cf = new LettuceConnectionFactory("localhost", 6379);