test(stream): 补 autoclaim 多实例并发幂等真 Redis 集成测试

两个 consumerName 不同的 StreamAutoClaimer 用 CyclicBarrier 同时对同一条 idle 超时的
PEL 消息发 XCLAIM,断言业务只被处理一次、PEL 清零。端到端覆盖 mock 单测与单实例死信
集成测试都证明不了的真实并发竞态(XCLAIM 归属转移 + idle 重置由 Redis 串行裁判),为
B1「先抢占再处理」补上端到端铁证。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/test/java/com/njcn/middle/stream/container/StreamAutoClaimConcurrentClaimTest.java

@@ -0,0 +1,231 @@
+package com.njcn.middle.stream.container;
+
+import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
+import com.njcn.middle.stream.domain.StreamBaseMessage;
+import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler;
+import com.njcn.middle.stream.support.StreamKeyBuilder;
+import com.njcn.middle.stream.template.RedisStreamEnhanceTemplate;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ConditionEvaluationResult;
+import org.junit.jupiter.api.extension.ExecutionCondition;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.data.domain.Range;
+import org.springframework.data.redis.connection.stream.Consumer;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.PendingMessages;
+import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.StreamOffset;
+import org.springframework.data.redis.connection.stream.StreamReadOptions;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * 集成测试（Gate）：autoclaim 多实例并发认领同一条 idle 超时的 PEL 消息时，业务必须只被处理一次。
+ *
+ * <p>端到端验证 B1 幂等：两个 consumerName 不同的 {@link StreamAutoClaimer} 用 {@link CyclicBarrier}
+ * 同时对同一条消息发 {@code XCLAIM}（不带 JUSTID）。Redis 单线程串行处理命令——先到者认领成功、把
+ * idle 重置为 0；后到者看到 {@code idle≈0 < minIdle} 必然抢不到（{@link StreamAutoClaimer#claim}
+ * 返回 false）→ 直接退出，不 dispatch。故无论哪个线程先到，{@code handleMessage} 恒只触发一次。
+ *
+ * <p>这条用例覆盖纯 mock 单测（{@code StreamAutoClaimerClaimTest}）和单实例死信集成测试都证明不了的
+ * <b>真实并发竞态</b>：mock 短路了 XCLAIM 的真实归属转移 + idle 重置语义，唯有真 Redis 能裁判。
+ *
+ * <p>测试放在 {@code container} 包以调用 package-private 的 {@link StreamAutoClaimer#claimOne}，
+ * 直接复现 autoclaim 单条认领，不依赖定时器时序，结果确定（非 flaky）。
+ *
+ * <p>Redis 不可达时整类禁用（skipped）。连接默认 {@code 127.0.0.1:6379}，用
+ * {@code -Dspring.redis.host=… -Dspring.redis.port=…} 覆盖。本地起：{@code docker run -d -p 6379:6379 redis:6.2}。
+ */
+@ExtendWith(StreamAutoClaimConcurrentClaimTest.RedisReachableCondition.class)
+@SpringBootTest(
+        classes = StreamAutoClaimConcurrentClaimTest.TestApp.class,
+        properties = {
+                "redis-stream.env-isolation=false",
+                "redis-stream.read-count=10"
+        })
+class StreamAutoClaimConcurrentClaimTest {
+
+    private static final String TOPIC = "IT_Concurrent_Topic";
+    private static final String GROUP = "IT_Concurrent_Group";
+    private static final long MIN_IDLE_MS = 500L;
+
+    @Autowired
+    private RedisStreamEnhanceTemplate template;
+
+    @Autowired
+    private StringRedisTemplate redis;
+
+    @Autowired
+    private StreamKeyBuilder keyBuilder;
+
+    @Test
+    void onlyOneInstanceProcessesUnderConcurrentClaim() throws Exception {
+        CountingHandler handler = new CountingHandler();
+        String stream = keyBuilder.buildStream(TOPIC);
+
+        // 清历史 + 重建组（delete 会连带删组，XADD 不重建，故须手动 MKSTREAM 建组）。
+        redis.delete(stream);
+        createGroup(stream, GROUP);
+
+        // 发一条真实消息（含 enc/body，可被 dispatchMessage 正常 decode→反序列化）。
+        ConcurrentMsg msg = new ConcurrentMsg();
+        msg.setText("process-once");
+        template.send(TOPIC, msg, false);
+
+        // 用 origin consumer XREADGROUP 读进 PEL（不 ACK），投递计数=1；之后 idle 从 0 起涨。
+        List<MapRecord<String, Object, Object>> recs = redis.opsForStream().read(
+                Consumer.from(GROUP, "origin-consumer"),
+                StreamReadOptions.empty().count(10),
+                StreamOffset.create(stream, ReadOffset.lastConsumed()));
+        assertEquals(1, recs == null ? 0 : recs.size(), "应有 1 条进入 PEL");
+        String id = recs.get(0).getId().getValue();
+
+        // 等 idle 超过 minIdle，使首个认领者满足 XCLAIM 的 min-idle 门槛。
+        Thread.sleep(MIN_IDLE_MS + 300L);
+
+        StreamAutoClaimer claimerA = startedClaimer("concurrent-claimer-A", handler);
+        StreamAutoClaimer claimerB = startedClaimer("concurrent-claimer-B", handler);
+        try {
+            CyclicBarrier gate = new CyclicBarrier(2);
+            ExecutorService pool = Executors.newFixedThreadPool(2);
+            List<Future<Void>> futures = pool.invokeAll(Arrays.asList(
+                    claimTask(gate, claimerA, handler, stream, id),
+                    claimTask(gate, claimerB, handler, stream, id)));
+            for (Future<Void> f : futures) {
+                f.get(10, TimeUnit.SECONDS); // 解包任务内异常，使其转为测试失败
+            }
+            pool.shutdownNow();
+        } finally {
+            claimerA.stop();
+            claimerB.stop();
+        }
+
+        assertEquals(1, handler.handled.get(),
+                "两个实例并发认领同一条，业务只应被处理一次（多实例幂等）");
+
+        PendingMessages pel = redis.opsForStream().pending(stream, GROUP, Range.unbounded(), 100);
+        assertEquals(0L, pel == null ? 0 : pel.size(), "认领方 ACK 后 PEL 应清零");
+    }
+
+    /** 两个 claimer 用 deliveries=1（< maxRetry）走正常 dispatch 分支；barrier 让二者尽量同时发 XCLAIM。 */
+    private Callable<Void> claimTask(CyclicBarrier gate, StreamAutoClaimer claimer,
+                                     CountingHandler handler, String stream, String id) {
+        return () -> {
+            gate.await();
+            claimer.claimOne(handler, stream, GROUP, MIN_IDLE_MS, 100, id, 1);
+            return null;
+        };
+    }
+
+    /** new + start()：start() 初始化 claimConsumer；interval 设大，定时器在测试窗内不触发，避免干扰。 */
+    private StreamAutoClaimer startedClaimer(String consumerName, CountingHandler handler) {
+        RedisStreamProperties props = new RedisStreamProperties();
+        props.setConsumerName(consumerName);
+        props.getAutoclaim().setIntervalMs(600000);
+        StreamAutoClaimer claimer = new StreamAutoClaimer(
+                redis, keyBuilder, props, Collections.singletonList(handler));
+        claimer.start();
+        return claimer;
+    }
+
+    private void createGroup(String stream, String group) {
+        redis.execute((RedisCallback<Object>) conn -> {
+            try {
+                return conn.execute("XGROUP",
+                        "CREATE".getBytes(StandardCharsets.UTF_8),
+                        stream.getBytes(StandardCharsets.UTF_8),
+                        group.getBytes(StandardCharsets.UTF_8),
+                        "$".getBytes(StandardCharsets.UTF_8),
+                        "MKSTREAM".getBytes(StandardCharsets.UTF_8));
+            } catch (RuntimeException e) {
+                return null; // BUSYGROUP：组已存在，忽略
+            }
+        });
+    }
+
+    /** 不注册任何 handler bean：autoconfig 的 listener/autoclaimer 因 handlers 为空而空转，不干扰本测试。 */
+    @SpringBootConfiguration
+    @EnableAutoConfiguration
+    static class TestApp {
+    }
+
+    @Data
+    @EqualsAndHashCode(callSuper = true)
+    static class ConcurrentMsg extends StreamBaseMessage {
+        private String text;
+    }
+
+    /** 真实 dispatch 路径计数（decode→反序列化→handleMessage），统计业务被处理的次数。 */
+    static class CountingHandler extends EnhanceStreamConsumerHandler<ConcurrentMsg> {
+        final AtomicInteger handled = new AtomicInteger();
+
+        @Override
+        public String topic() {
+            return TOPIC;
+        }
+
+        @Override
+        public String group() {
+            return GROUP;
+        }
+
+        @Override
+        public Class<ConcurrentMsg> messageType() {
+            return ConcurrentMsg.class;
+        }
+
+        @Override
+        public void handleMessage(ConcurrentMsg message) {
+            handled.incrementAndGet();
+        }
+
+        @Override
+        public boolean isRetry() {
+            return true;
+        }
+
+        @Override
+        public boolean throwException() {
+            return false;
+        }
+    }
+
+    /** Redis 不可达时禁用整个测试类（上下文加载即连 Redis，故须在加载前判断）。 */
+    static class RedisReachableCondition implements ExecutionCondition {
+        @Override
+        public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context) {
+            String host = System.getProperty("spring.redis.host", "127.0.0.1");
+            int port = Integer.parseInt(System.getProperty("spring.redis.port", "6379"));
+            try (Socket s = new Socket()) {
+                s.connect(new InetSocketAddress(host, port), 500);
+                return ConditionEvaluationResult.enabled("redis reachable at " + host + ":" + port);
+            } catch (Exception e) {
+                return ConditionEvaluationResult.disabled(
+                        "no redis at " + host + ":" + port + ", skip integration test");
+            }
+        }
+    }
+}