fix: reset the claim cursor whenever the consumer group is recreated

recoverFromNoGroup is reached from both readMessages and
claimPendingMessages, but the cursor reset sat at the claim call site only.
The read path is the likelier of the two to notice NOGROUP first, since reads
run on every tick that is not a claim tick -- so the common case left the
cursor pointing into the pending list of a group that no longer exists.

Not self-correcting on a busy stream. A scan starting above the recreated
PEL's entries only wraps once it exhausts the list, and entries arriving after
the stale position keep giving it work at the high end, so the wrap can be
deferred indefinitely while the oldest entries go unexamined -- the exact
starvation this PR exists to remove.

Reset moved into recoverFromNoGroup so both paths get identical treatment and
there is one place that owns it.

Regression test covers the read path specifically. Mutation-checked: moving
the reset back to the claim site alone fails it.

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

Author: thiagohora

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriber.java

@@ -434,10 +434,9 @@ private Mono<Map<StreamMessageId, Map<String, M>>> claimPendingMessages() {
                     claimErrors.add(1);
                     log.error("Error claiming pending messages", throwable);
                     // A failed scan leaves the cursor where it was, so the next attempt retries the same
-                    // window rather than skipping it. Except on NOGROUP: the group is being recreated, so
-                    // any cursor into the old group's PEL is meaningless.
+                    // window rather than skipping it. The NOGROUP case is different, but recoverFromNoGroup
+                    // resets the cursor itself so both it and the read path get the same treatment.
                     if (isNoGroupError(throwable)) {
-                        claimCursor = StreamMessageId.MIN;
                         return recoverFromNoGroup();
                     }
                     return Mono.just(Map.of());
@@ -507,6 +506,16 @@ private boolean isNoGroupError(Throwable throwable) {
     private Mono<Map<StreamMessageId, Map<String, M>>> recoverFromNoGroup() {
         log.warn("Recreating not found consumer group '{}' for stream '{}'",
                 config.getConsumerGroupName(), config.getStreamName());
+        // The cursor indexes the OLD group's pending list, so it means nothing once the group is
+        // recreated. Reset here rather than at the call sites: NOGROUP surfaces from both readMessages
+        // and claimPendingMessages, and the read path is the likelier of the two to notice it first
+        // (reads run on every tick that is not a claim tick).
+        //
+        // Leaving it stale is not self-correcting on a busy stream. A scan starting above the recreated
+        // PEL's entries only wraps once it exhausts the list, and entries arriving after the stale
+        // position keep giving it work at the high end -- so the wrap can be deferred indefinitely while
+        // the oldest entries, the ones this scan exists to reach, are never examined.
+        claimCursor = StreamMessageId.MIN;
         return createConsumerGroup()
                 .onErrorResume(throwable -> {
                     log.error("Failed to recreate consumer group '{}' for stream '{}'",

apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriberUnitTest.java

@@ -622,6 +622,58 @@ void shouldHandleInvalidMessageIdTimestamp() {
     @Nested
     class NoGroupErrorTests {
 
+        /**
+         * Regression for review feedback on OPIK-8240: {@code recoverFromNoGroup} is reached from BOTH
+         * {@code readMessages} and {@code claimPendingMessages}, and the cursor reset was originally only
+         * at the claim call site. The read path is the likelier of the two to notice NOGROUP first, since
+         * reads run on every tick that is not a claim tick.
+         *
+         * <p>A cursor left pointing into the old group's pending list is not self-correcting on a busy
+         * stream: a scan starting above the recreated PEL's entries only wraps once it exhausts the list,
+         * and entries arriving after the stale position keep feeding it at the high end, so the oldest
+         * entries can go unexamined indefinitely -- the exact starvation this fix exists to remove.
+         */
+        @Test
+        void shouldResetTheClaimCursorWhenTheGroupIsRecreatedViaTheReadPath() {
+            whenCreateGroupReturnEmpty();
+            whenRemoveConsumerReturn();
+            var fastConfig = CONFIG.toBuilder().claimIntervalRatio(2).build();
+            var subscriber = trackSubscriber(TestRedisSubscriber.createSubscriber(fastConfig, redissonClient));
+            var starts = new CopyOnWriteArrayList<StreamMessageId>();
+            var claims = new AtomicInteger();
+
+            // First claim advances the cursor well past the start of the PEL.
+            when(stream.autoClaim(
+                    eq(fastConfig.getConsumerGroupName()),
+                    anyString(),
+                    eq(fastConfig.getPendingMessageDuration().toJavaDuration().toMillis()),
+                    eq(TimeUnit.MILLISECONDS),
+                    any(StreamMessageId.class),
+                    eq(fastConfig.getConsumerBatchSize())))
+                    .thenAnswer(invocation -> {
+                        starts.add(invocation.getArgument(4));
+                        claims.incrementAndGet();
+                        return Mono.just(new AutoClaimResult<>(
+                                new StreamMessageId(9_000L, 0), Map.of(), List.of()));
+                    });
+            // Reads then hit NOGROUP, which recreates the group underneath us.
+            when(stream.readGroup(eq(fastConfig.getConsumerGroupName()), anyString(),
+                    any(StreamReadGroupArgs.class)))
+                    .thenAnswer(invocation -> Mono
+                            .error(new RuntimeException("NOGROUP No such key stream or consumer group")));
+            whenAckReturn();
+            whenRemoveReturn();
+
+            subscriber.start();
+
+            await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).until(() -> starts.size() >= 2);
+
+            // The recreated group has a fresh PEL, so the next scan must start from the beginning rather
+            // than from the position it reached in the group that no longer exists.
+            assertThat(starts.getFirst()).isEqualTo(StreamMessageId.MIN);
+            assertThat(starts.get(1)).isEqualTo(StreamMessageId.MIN);
+        }
+
         @Test
         void shouldRecoverOnClaimAndNotDie() {
             whenCreateGroupReturnEmpty();