Using the KEDA scaler with Redis Streams: Ideal Usage Question

[nakumgaurav]

[Continuing the discussion from #746 ]

Problem:
If I have an application adding messages to a stream, they are not delivered to any consumer group until an XREADGROUP is issued over that consumer group. This means that if the Job that KEDA is trying to scale consumes messages, it has to wait for someone to consume messages (!!!) using XREADGROUP so that the messages are moved to the Pending Entries List (PEL) of that group. This leaves Redis Streams pretty much useless with KEDA because if some other application has already fetched the messages, how would I retrieve them in my KEDA application that I am trying to scale?

I guess this is more related to how Redis consumer groups work but if there is a KEDA way of doing things, would love to hear that.

Potential Workaround:
What I ended up doing was:

1. In the producer application, get a list of all the consumer groups over a given stream using XINFO GROUPS
2. Use XREADGROUP to deliver messages to all the groups
3. In the consumer application (the job that KEDA is scaling), obtained the range of ids in the PEL of the desired group - over which the KEDA trigger is defined - using XPENDING and read the messages in this range using XRANGE
4. After processing the messages in the consumer app, acknowledge them using XACK so that they are removed from the PEL of the group and more messages could be received.

I realize this may not be the ideal way to solve the problem, so open to suggestions. Thank you!

[KomTak001]

I think, we need count of not delivered entry count for auto scaling metrics like this.
How about this?

• retrieve last-delivered-id for specific consumer group

XINFO GROUPS [streameName]
  :
name : "some-group-1"
last-delivered-id : "1638125141232-0"
  : 
  :
name : "some-group-2"
last-delivered-id : "1638125140123-0"
  : 

• retrieve not delivered entry count, and compare threshold (O(N) N=threashold)

XRANGE [streameName] [last-derivered-id for specific group] + COUNT [threashold]
  (entry list)