Add support for XREADGROUP CLAIM arg #3486

a-TODO-rov · 2025-10-23T14:11:43Z

Copilot

Pull Request Overview

This pull request adds support for Redis XREADGROUP CLAIM functionality, allowing the claiming of idle pending messages from a stream's Pending Entries List (PEL) with a configurable minimum idle time. The implementation introduces delivery metadata tracking for claimed entries.

Adds XReadArgs.claim() methods to specify minimum idle time for claiming entries
Introduces ClaimedStreamMessage class to carry metadata (milliseconds since last delivery and redelivery count)
Updates StreamReadOutput to parse optional metadata fields from Redis responses
Adds comprehensive unit and integration tests

Reviewed Changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 16 comments.

Show a summary per file

File	Description
src/main/java/io/lettuce/core/XReadArgs.java	Adds `claim()` builder and instance methods to configure CLAIM min-idle-time parameter
src/main/java/io/lettuce/core/StreamMessage.java	Adds `isClaimed()` method to distinguish claimed entries from fresh ones
src/main/java/io/lettuce/core/ClaimedStreamMessage.java	New class extending StreamMessage to hold claim metadata (redelivery count and time since last delivery)
src/main/java/io/lettuce/core/output/StreamReadOutput.java	Updates output parser to handle optional metadata fields for claimed entries in both integer and bulk string formats
src/test/java/io/lettuce/core/output/StreamReadOutputUnitTests.java	Adds unit tests for parsing claimed entries with metadata in various formats
src/test/java/io/lettuce/core/XReadGroupClaimIntegrationTests.java	New integration test suite validating CLAIM behavior with real Redis server

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-11-06T19:04:16Z

src/main/java/io/lettuce/core/output/StreamReadOutput.java

+            if (redeliveryCount == null && isDigits(s)) {
+                redeliveryCount = Long.parseLong(s);
+                return;
+            }


Potential logic issue: If a non-numeric bulk string is received when extras are expected (id != null && bodyReceived && key == null), the code will skip the extras handling (lines 64-71) because isDigits(s) returns false, but then fall through to line 79 where it would incorrectly interpret this data as a body key. This could cause incorrect parsing.

Consider restructuring the logic to handle unexpected data more explicitly, or add an else clause to log/handle non-numeric extras:

if (id != null && bodyReceived && key == null && bytes != null) { String s = decodeString(bytes.duplicate()); if (msSinceLastDelivery == null && isDigits(s)) { msSinceLastDelivery = Long.parseLong(s); return; } if (redeliveryCount == null && isDigits(s)) { redeliveryCount = Long.parseLong(s); return; } // If we reach here with non-numeric data in the extras position, // it should be handled or logged, not allowed to fall through }

Suggested change

}

}

// If we reach here, non-numeric data was received in the extras position.

// Log or handle the unexpected data. For now, we ignore and do not fall through.

// System.err.println("Unexpected non-numeric extras data: " + s);

return;

Copilot · 2025-11-06T19:04:17Z

src/test/java/io/lettuce/core/XReadGroupClaimIntegrationTests.java

+            sync.xadd(key, body);
+            sync.xadd(key, body);
+
+            // First read with CLAIM(minIdle=100ms); entries should be fresh (not claimed)


Comment is inconsistent with the code. The comment states "minIdle=100ms" but the code uses Duration.ofMillis(50). Please update the comment to match the actual value:

// First read with CLAIM(minIdle=50ms); entries should be fresh (not claimed)

Suggested change

// First read with CLAIM(minIdle=100ms); entries should be fresh (not claimed)

// First read with CLAIM(minIdle=50ms); entries should be fresh (not claimed)

Copilot · 2025-11-06T19:04:17Z

src/main/java/io/lettuce/core/XReadArgs.java


+    /**
+     * Claim idle pending messages first with a minimum idle time (milliseconds). Only valid for XREADGROUP.
+     * 


Missing JavaDoc parameters and return value documentation. The method should document:

@param milliseconds the minimum idle time in milliseconds

@return {@code this} XReadArgs instance for method chaining

Suggested change

*

*

* @param milliseconds the minimum idle time in milliseconds

* @return {@code this} XReadArgs instance for method chaining

Copilot · 2025-11-06T19:04:18Z

src/main/java/io/lettuce/core/XReadArgs.java

+
+    /**
+     * Claim idle pending messages first with a minimum idle time. Only valid for XREADGROUP.
+     * 


Missing JavaDoc parameters and return value documentation. The method should document:

@param timeout the minimum idle time as a Duration

@return {@code this} XReadArgs instance for method chaining

Suggested change

*

*

* @param timeout the minimum idle time as a Duration

* @return {@code this} XReadArgs instance for method chaining

Copilot · 2025-11-06T19:04:18Z

src/main/java/io/lettuce/core/XReadArgs.java

        }

+        /**
+         * Create a new {@link XReadArgs} and set CLAIM min-idle-time (milliseconds). Only valid for XREADGROUP.


Missing JavaDoc parameters and return value documentation. The method should document:

@param milliseconds the minimum idle time in milliseconds

@return new {@link XReadArgs} with CLAIM set

Consider adding @see XReadArgs#claim(long) for consistency with other Builder methods

Suggested change

* Create a new {@link XReadArgs} and set CLAIM min-idle-time (milliseconds). Only valid for XREADGROUP.

* Create a new {@link XReadArgs} and set CLAIM min-idle-time (milliseconds). Only valid for XREADGROUP.

*

* @param milliseconds the minimum idle time in milliseconds

* @return new {@link XReadArgs} with CLAIM set

* @see XReadArgs#claim(long)

Copilot · 2025-11-06T19:04:20Z

src/main/java/io/lettuce/core/ClaimedStreamMessage.java

+    public long getMsSinceLastDelivery() {
+        return msSinceLastDelivery;
+    }
+


Missing JavaDoc for the public method. Should include:

A description of what the method returns

@return Duration since the last delivery of this message

Suggested change

/**

* Returns the duration since the last delivery of this message.

*

* @return Duration since the last delivery of this message.

*/

src/main/java/io/lettuce/core/StreamMessage.java

Copilot · 2025-11-06T19:04:21Z

src/main/java/io/lettuce/core/output/StreamReadOutput.java

+                msSinceLastDelivery = Long.parseLong(s);
+                return;
+            }
+            if (redeliveryCount == null && isDigits(s)) {
+                redeliveryCount = Long.parseLong(s);


Potential uncaught 'java.lang.NumberFormatException'.

Suggested change

msSinceLastDelivery = Long.parseLong(s);

return;

}

if (redeliveryCount == null && isDigits(s)) {

redeliveryCount = Long.parseLong(s);

try {

msSinceLastDelivery = Long.parseLong(s);

} catch (NumberFormatException e) {

msSinceLastDelivery = null; // or handle as appropriate

}

return;

}

if (redeliveryCount == null && isDigits(s)) {

try {

redeliveryCount = Long.parseLong(s);

} catch (NumberFormatException e) {

redeliveryCount = null; // or handle as appropriate

}

Copilot · 2025-11-06T19:04:21Z

src/main/java/io/lettuce/core/output/StreamReadOutput.java

+                msSinceLastDelivery = Long.parseLong(s);
+                return;
+            }
+            if (redeliveryCount == null && isDigits(s)) {
+                redeliveryCount = Long.parseLong(s);


Potential uncaught 'java.lang.NumberFormatException'.

Suggested change

msSinceLastDelivery = Long.parseLong(s);

return;

}

if (redeliveryCount == null && isDigits(s)) {

redeliveryCount = Long.parseLong(s);

try {

msSinceLastDelivery = Long.parseLong(s);

} catch (NumberFormatException e) {

// Optionally log or handle the error here

}

return;

}

if (redeliveryCount == null && isDigits(s)) {

try {

redeliveryCount = Long.parseLong(s);

} catch (NumberFormatException e) {

// Optionally log or handle the error here

}

src/test/java/io/lettuce/core/XReadGroupClaimIntegrationTests.java

a-TODO-rov · 2025-11-06T19:58:58Z

src/main/java/io/lettuce/core/output/StreamReadOutput.java

+            }
+        }
+        super.set(integer);
+    }


Possibly unused code. Seems server sends string instead the specified integer.

a-TODO-rov requested a review from uglide October 23, 2025 14:13

a-TODO-rov linked an issue Oct 23, 2025 that may be closed by this pull request

Add support for CLAIM option on XREADGROUP #3482

Open

Add support for XREADGROUP CLAIM arg

afc4947

a-TODO-rov force-pushed the xreadgorup-claim branch from 19e2d4b to afc4947 Compare October 27, 2025 12:41

Merge branch 'main' into xreadgorup-claim

84438de

a-TODO-rov marked this pull request as ready for review November 3, 2025 14:54

a-TODO-rov added 2 commits November 4, 2025 11:16

Add NOACK scenario in ITs

2e04494

Fix NOACK IT scenario. Add test.

085d857

a-TODO-rov requested review from Copilot, ggivo and tishun and removed request for ggivo, tishun and uglide November 5, 2025 15:26

Copilot AI reviewed Nov 6, 2025

View reviewed changes

a-TODO-rov commented Nov 6, 2025

View reviewed changes

redis deleted a comment from Copilot AI Nov 7, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Add support for XREADGROUP CLAIM arg #3486

Add support for XREADGROUP CLAIM arg #3486

Uh oh!

a-TODO-rov commented Oct 23, 2025

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Nov 6, 2025

Uh oh!

Copilot AI Nov 6, 2025

Uh oh!

Copilot AI Nov 6, 2025

Uh oh!

Copilot AI Nov 6, 2025

Uh oh!

Copilot AI Nov 6, 2025

Uh oh!

Copilot AI Nov 6, 2025

Uh oh!

Uh oh!

Copilot AI Nov 6, 2025

Uh oh!

Copilot AI Nov 6, 2025

Uh oh!

Uh oh!

a-TODO-rov Nov 6, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

-            }
+            }
+            // If we reach here, non-numeric data was received in the extras position.
+            // Log or handle the unexpected data. For now, we ignore and do not fall through.
+            // System.err.println("Unexpected non-numeric extras data: " + s);
+            return;

	// First read with CLAIM(minIdle=100ms); entries should be fresh (not claimed)
	// First read with CLAIM(minIdle=50ms); entries should be fresh (not claimed)

-     *
+     *
+     * @param milliseconds the minimum idle time in milliseconds
+     * @return {@code this} XReadArgs instance for method chaining

+    /**
+     * Returns the duration since the last delivery of this message.
+     *
+     * @return Duration since the last delivery of this message.
+     */

Add support for XREADGROUP CLAIM arg #3486

Are you sure you want to change the base?

Add support for XREADGROUP CLAIM arg #3486

Uh oh!

Conversation

a-TODO-rov commented Oct 23, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI Nov 6, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 6, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 6, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 6, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 6, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 6, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Copilot AI Nov 6, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Nov 6, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

a-TODO-rov Nov 6, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants