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>
fix(stream): autoclaim 先 XCLAIM 抢占再处理,杜绝多实例重复消费/重复死信
2026-06-26 14:33:29 +08:00 · 2026-06-26 14:21:38 +08:00 · 2026-06-26 11:41:56 +08:00
5 changed files with 475 additions and 11 deletions
--- a/src/main/java/com/njcn/middle/stream/autoconfig/RedisStreamAutoConfiguration.java
+++ b/src/main/java/com/njcn/middle/stream/autoconfig/RedisStreamAutoConfiguration.java
@@ -8,6 +8,7 @@ import com.njcn.middle.stream.support.StreamTrimScheduler;
 import com.njcn.middle.stream.template.RedisStreamEnhanceTemplate;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
@@ -26,8 +27,9 @@ import java.util.stream.Collectors;
 * Phase 2 由 message-boot 注入 Processor 即生效。装配在 {@link RedisAutoConfiguration} 之后，
 * 确保 {@link StringRedisTemplate} 已就绪。
 */
-@Configuration
+@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(StringRedisTemplate.class)
+@ConditionalOnBean(StringRedisTemplate.class)
@AutoConfigureAfter(RedisAutoConfiguration.class)
@EnableConfigurationProperties(RedisStreamProperties.class)
 public class RedisStreamAutoConfiguration {
--- a/src/main/java/com/njcn/middle/stream/container/StreamAutoClaimer.java
+++ b/src/main/java/com/njcn/middle/stream/container/StreamAutoClaimer.java
@@ -17,6 +17,7 @@ import org.springframework.data.redis.core.RedisCallback;
 import org.springframework.data.redis.core.StringRedisTemplate;

 import java.nio.charset.StandardCharsets;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -166,15 +167,18 @@ public class StreamAutoClaimer implements SmartLifecycle {
    }

    /**
-     * 处理单条 PEL 消息：取内容 → 死信判定前置 → 认领重投。
+     * 处理单条 PEL 消息：取内容 → XCLAIM 抢占 → 死信判定 → 认领重投。
     *
-     * <p>{@code deliveries} 为本轮 {@code XCLAIM} 之前的投递计数；因 {@link #claim} 用<b>不带 JUSTID</b>
-     * 的 XCLAIM，每次认领会把投递计数 +1，故 deliveries 随 autoclaim 轮次单调增长。
+     * <p><b>先抢占再处理</b>：{@link #claim} 抢不到（idle 不足 minIdle，或并发实例已认领并重置 idle）
+     * 即直接退出,保证多实例下同一条消息只被一个实例 dispatch / 落死信(幂等),不重复消费、不重复落死信。
+     *
+     * <p>{@code deliveries} 为本轮 {@code XCLAIM} 之前的投递计数（XPENDING 报告值）；因 {@link #claim}
+     * 用<b>不带 JUSTID</b> 的 XCLAIM，每次认领会把投递计数 +1，故 deliveries 随 autoclaim 轮次单调增长。
     * {@code maxRetry} 语义 = 最多投递 maxRetry 次（含首投），第 maxRetry 次仍 RETRY 即落死信。
-     * 死信判定<b>前置于 dispatch</b>：达上限直接落异常 + 强制 ACK，不再多跑一次业务副作用。
+     * 死信分支抢占后<b>只落异常 + 强制 ACK，不再 dispatch</b>，不跑业务副作用。
     */
-    private void claimOne(EnhanceStreamConsumerHandler<?> handler, String stream, String group,
-                          long minIdle, int maxRetry, String id, long deliveries) {
+    void claimOne(EnhanceStreamConsumerHandler<?> handler, String stream, String group,
+                  long minIdle, int maxRetry, String id, long deliveries) {
        List<MapRecord<String, Object, Object>> records =
                redis.opsForStream().range(stream, Range.closed(id, id));
        if (records == null || records.isEmpty()) {
@@ -185,6 +189,12 @@ public class StreamAutoClaimer implements SmartLifecycle {
        MapRecord<String, Object, Object> rec = records.get(0);
        Map<String, String> fields = toStringMap(rec.getValue());

+        // 先 XCLAIM 抢占：仅当真正认领到（idle 足够且未被并发实例抢走）才由本实例独占处理；
+        // 抢不到说明已有其它实例认领，直接退出——杜绝多实例对同一条消息重复 dispatch 或重复落死信。
+        if (!claim(stream, group, minIdle, id)) {
+            return;
+        }
+
        if (deliveries >= maxRetry) {
            // 投递计数已达上限 → 落异常 + 强制 ACK（死信，不再无限滞留 PEL）
            // 由 handler 内部反序列化后回调,使死信回调能拿到业务消息对象（解析失败兜底 null）
@@ -194,8 +204,7 @@ public class StreamAutoClaimer implements SmartLifecycle {
            return;
        }

-        // 认领到本 claimer（同时把投递计数 +1、重置 idle），再重投
-        claim(stream, group, minIdle, id);
+        // 已认领（投递计数 +1、idle 重置）→ 重投
        ConsumeResult result = handler.dispatchMessage(fields);
        if (result == ConsumeResult.ACK) {
            redis.opsForStream().acknowledge(stream, group, rec.getId());
@@ -209,14 +218,24 @@ public class StreamAutoClaimer implements SmartLifecycle {
     * <p>不带 JUSTID 时 XCLAIM 会转移归属、重置 idle 并把投递计数 +1——后者是 maxRetry 死信判定的
     * 依据：listener 用 {@code XREADGROUP lastConsumed()} 永不重读 PEL，若此处用 JUSTID 则投递计数
     * 永久冻结在首投值 1，{@code deliveries >= maxRetry} 恒不成立、死信分支永不触发。
+     *
+     * @return 是否真正认领到本条：XCLAIM 抢到返回非空消息列表，没抢到返回空。返回类型未知时保守当作
+     *         抢到（宁可退化回"可能重复"也不让消息因误判没抢到而永久卡 PEL）。
     */
-    private void claim(String stream, String group, long minIdle, String id) {
-        redis.execute((RedisCallback<Object>) connection -> connection.execute("XCLAIM",
+    private boolean claim(String stream, String group, long minIdle, String id) {
+        Object claimed = redis.execute((RedisCallback<Object>) connection -> connection.execute("XCLAIM",
                stream.getBytes(StandardCharsets.UTF_8),
                group.getBytes(StandardCharsets.UTF_8),
                claimConsumer.getBytes(StandardCharsets.UTF_8),
                String.valueOf(minIdle).getBytes(StandardCharsets.UTF_8),
                id.getBytes(StandardCharsets.UTF_8)));
+        if (claimed instanceof Collection) {
+            return !((Collection<?>) claimed).isEmpty();
+        }
+        if (claimed instanceof Object[]) {
+            return ((Object[]) claimed).length > 0;
+        }
+        return claimed != null;
    }

    private static Map<String, String> toStringMap(Map<Object, Object> raw) {
--- a/src/test/java/com/njcn/middle/stream/autoconfig/RedisStreamAutoConfigurationTest.java
+++ b/src/test/java/com/njcn/middle/stream/autoconfig/RedisStreamAutoConfigurationTest.java
@@ -0,0 +1,71 @@
+package com.njcn.middle.stream.autoconfig;
+
+import com.njcn.middle.stream.container.StreamAutoClaimer;
+import com.njcn.middle.stream.container.StreamListenerContainer;
+import com.njcn.middle.stream.support.StreamKeyBuilder;
+import com.njcn.middle.stream.support.StreamTrimScheduler;
+import com.njcn.middle.stream.template.RedisStreamEnhanceTemplate;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+/**
+ * 自动装配测试:验证 stream starter 的 Bean 装配、用户自定义回退、以及缺 {@link StringRedisTemplate}
+ * 时整类优雅退场（与 cache 装配口径对齐,不因缺依赖启动失败）。
+ */
+class RedisStreamAutoConfigurationTest {
+
+    private final ApplicationContextRunner runner = new ApplicationContextRunner()
+            .withConfiguration(AutoConfigurations.of(
+                    RedisAutoConfiguration.class, RedisStreamAutoConfiguration.class))
+            .withPropertyValues("spring.redis.host=127.0.0.1", "spring.redis.port=6379");
+
+    @Test
+    void registersStreamBeans() {
+        runner.run(ctx -> {
+            assertThat(ctx).hasSingleBean(StreamKeyBuilder.class);
+            assertThat(ctx).hasSingleBean(RedisStreamEnhanceTemplate.class);
+            assertThat(ctx).hasSingleBean(StreamListenerContainer.class);
+            assertThat(ctx).hasSingleBean(StreamAutoClaimer.class);
+            assertThat(ctx).hasSingleBean(StreamTrimScheduler.class);
+        });
+    }
+
+    @Test
+    void backsOffWhenUserDefinesOwnTemplate() {
+        runner.withUserConfiguration(CustomConfig.class).run(ctx -> {
+            assertThat(ctx).hasSingleBean(RedisStreamEnhanceTemplate.class);
+            assertThat(ctx.getBean(RedisStreamEnhanceTemplate.class))
+                    .isSameAs(ctx.getBean("myTemplate"));
+        });
+    }
+
+    @Test
+    void backsOffWhenNoStringRedisTemplateBean() {
+        // 不装配 RedisAutoConfiguration,上下文内无 StringRedisTemplate Bean:
+        // 应整类 back off 优雅退场,而非因缺依赖启动失败（与 cache 装配口径一致）。
+        new ApplicationContextRunner()
+                .withConfiguration(AutoConfigurations.of(RedisStreamAutoConfiguration.class))
+                .run(ctx -> assertThat(ctx)
+                        .hasNotFailed()
+                        .doesNotHaveBean(RedisStreamEnhanceTemplate.class));
+    }
+
+    @Configuration
+    static class CustomConfig {
+        @Bean
+        RedisStreamEnhanceTemplate myTemplate() {
+            return new RedisStreamEnhanceTemplate(
+                    new StringRedisTemplate(mock(RedisConnectionFactory.class)),
+                    new StreamKeyBuilder(false, ""));
+        }
+    }
+}
--- a/src/test/java/com/njcn/middle/stream/container/StreamAutoClaimConcurrentClaimTest.java
+++ b/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");
+            }
+        }
+    }
+}
--- a/src/test/java/com/njcn/middle/stream/container/StreamAutoClaimerClaimTest.java
+++ b/src/test/java/com/njcn/middle/stream/container/StreamAutoClaimerClaimTest.java
@@ -0,0 +1,141 @@
+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.ConsumeResult;
+import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler;
+import com.njcn.middle.stream.support.StreamKeyBuilder;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.domain.Range;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.RecordId;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.StreamOperations;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * 单元测试:autoclaim 认领单条 PEL 消息（{@link StreamAutoClaimer#claimOne}）的多实例幂等。
+ *
+ * <p>核心:XCLAIM 真正抢到（返回非空）才由本实例 dispatch / 落死信;抢不到（并发实例已认领、
+ * idle 被重置)直接退出,杜绝重复消费与重复死信回调。
+ */
+class StreamAutoClaimerClaimTest {
+
+    private static final String STREAM = "T";
+    private static final String GROUP = "G";
+    private static final String ID = "1-0";
+
+    private final StringRedisTemplate redis = mock(StringRedisTemplate.class);
+    @SuppressWarnings("unchecked")
+    private final StreamOperations<String, Object, Object> ops = mock(StreamOperations.class);
+
+    @SuppressWarnings("unchecked")
+    private StreamAutoClaimer newClaimer(RecordingHandler handler) {
+        when(redis.opsForStream()).thenReturn(ops);
+        Map<Object, Object> fields = new HashMap<>();
+        fields.put("enc", "plain");
+        fields.put("body", "{}");
+        MapRecord<String, Object, Object> rec = mock(MapRecord.class);
+        when(rec.getValue()).thenReturn(fields);
+        when(rec.getId()).thenReturn(RecordId.of(ID));
+        when(ops.range(eq(STREAM), any(Range.class))).thenReturn(Collections.singletonList(rec));
+        return new StreamAutoClaimer(redis, new StreamKeyBuilder(false, ""),
+                new RedisStreamProperties(), Collections.singletonList(handler));
+    }
+
+    @SuppressWarnings("unchecked")
+    private void stubClaimAcquired(boolean acquired) {
+        when(redis.execute(any(RedisCallback.class)))
+                .thenReturn(acquired ? Collections.singletonList(new Object()) : Collections.emptyList());
+    }
+
+    @Test
+    void skipsDispatchWhenClaimLost() {
+        RecordingHandler handler = new RecordingHandler();
+        StreamAutoClaimer claimer = newClaimer(handler);
+        stubClaimAcquired(false); // 并发实例已抢走，本实例 XCLAIM 抢不到
+
+        claimer.claimOne(handler, STREAM, GROUP, 1000L, 3, ID, 0);
+
+        assertFalse(handler.dispatched, "未抢到 XCLAIM 不应 dispatch");
+        verify(ops, never()).acknowledge(eq(STREAM), eq(GROUP), any(RecordId.class));
+    }
+
+    @Test
+    void skipsDeadLetterWhenClaimLost() {
+        RecordingHandler handler = new RecordingHandler();
+        StreamAutoClaimer claimer = newClaimer(handler);
+        stubClaimAcquired(false);
+
+        claimer.claimOne(handler, STREAM, GROUP, 1000L, 3, ID, 9); // deliveries 已超限
+
+        assertFalse(handler.deadLettered, "未抢到 XCLAIM 不应重复落死信");
+        verify(ops, never()).acknowledge(eq(STREAM), eq(GROUP), any(RecordId.class));
+    }
+
+    @Test
+    void dispatchesWhenClaimAcquired() {
+        RecordingHandler handler = new RecordingHandler();
+        StreamAutoClaimer claimer = newClaimer(handler);
+        stubClaimAcquired(true);
+
+        claimer.claimOne(handler, STREAM, GROUP, 1000L, 3, ID, 0);
+
+        assertTrue(handler.dispatched, "抢到 XCLAIM 应 dispatch");
+        verify(ops, times(1)).acknowledge(eq(STREAM), eq(GROUP), any(RecordId.class));
+    }
+
+    @Test
+    void deadLettersWhenClaimAcquiredAndOverLimit() {
+        RecordingHandler handler = new RecordingHandler();
+        StreamAutoClaimer claimer = newClaimer(handler);
+        stubClaimAcquired(true);
+
+        claimer.claimOne(handler, STREAM, GROUP, 1000L, 3, ID, 9);
+
+        assertTrue(handler.deadLettered, "抢到且超限应落死信");
+        assertFalse(handler.dispatched, "死信不应再 dispatch 业务");
+        verify(ops, times(1)).acknowledge(eq(STREAM), eq(GROUP), any(RecordId.class));
+    }
+
+    static class TestMsg extends StreamBaseMessage {
+    }
+
+    /** 记录 dispatch / 死信回调是否被触发的测试桩。 */
+    static class RecordingHandler extends EnhanceStreamConsumerHandler<TestMsg> {
+        boolean dispatched = false;
+        boolean deadLettered = false;
+
+        @Override public String topic() { return STREAM; }
+        @Override public String group() { return GROUP; }
+        @Override public Class<TestMsg> messageType() { return TestMsg.class; }
+        @Override public void handleMessage(TestMsg message) { }
+        @Override public boolean isRetry() { return true; }
+        @Override public boolean throwException() { return false; }
+
+        @Override
+        public ConsumeResult dispatchMessage(Map<String, String> fields) {
+            dispatched = true;
+            return ConsumeResult.ACK;
+        }
+
+        @Override
+        public void saveExceptionFromFields(Map<String, String> fields, Exception cause) {
+            deadLettered = true;
+        }
+    }
+}