Adding default ACK timeout for messages in Java Consumer Groups

[tarunannapareddy]

I am exploring Redis Streams' reliability as a message queue in various scenarios. I have created a consumer group with 2 consumers: consumer1 and consumer2. In the scenario where consumer1 reads 5 messages but fails to ACK due to reasons such as server crashes, I want the other consumer to start receiving those messages after a fixed timeout period. This is similar to how the ConsumerGroup concept works in Apache Kafka/Pulsar. Can I recreate this scenario with Redis? If not yet, Can I Contribute to adding this feature?

public static void main(String[] args) {

        RedisClient redisClient = RedisClient.create("redis://127.0.0.1:6379");
        StatefulRedisConnection<String, String> connection = redisClient.connect();
        RedisStreamCommands<String, String> streamCommands = connection.sync();
        String lastSeenOffset = "0-0";
        try {
            streamCommands.xgroupCreate( XReadArgs.StreamOffset.from(STREAM_KEY, "0-0"), CONSUMER_GROUP );
        }
        catch (RedisBusyException redisBusyException) {
            System.out.println( String.format("\t Group '%s' already exists",CONSUMER_GROUP));
        }
        while(true){
            List<StreamMessage<String, String>> messages = streamCommands.xreadgroup(Consumer.from(CONSUMER_GROUP, CONSUMER),
                    XReadArgs.StreamOffset.lastConsumed(STREAM_KEY)
            );
            if(!messages.isEmpty()){
                for(StreamMessage<String, String> message : messages){
                    System.out.print(lastSeenOffset+" received message");
                    System.out.print(" userId: " +message.getBody().get("userId"));
                    System.out.print(" order: "+message.getBody().get("order"));
                    System.out.println(" email: "+message.getBody().get("e-mail"));
                    streamCommands.xack(STREAM_KEY, CONSUMER_GROUP,  message.getId());
                }
            }
        }
    }

[mp911de]

Lettuce has intentionally no polling or retry functionality because each requirement regarding retries is different. The API focuses on pure Redis command interaction. Any higher-level functionality is subject to a higher-level library. It would be neat to have a general library for stream interaction built on Lettuce that handles these aspects.

[tarunannapareddy]

Hi @mp911de

Can you help me with the best way to write my code in the above situation? Should I use xClaim(with required pending timeout) every time along with xreadGroup in the while loop and manually combine both lists and use it? or is there any other library/function that provides both the features in one?

Also, I agree everyone has different retry policies. But I believe adding timed-out events to another consumer's pool is the most basic. It will help most of the users using consumer groups, others can just not add timeout in xgroupCreate and this base functionality still preserves.