feat(pubsub/redis): add streamStartID metadata option

The redis streams pubsub always created new consumer groups starting
from entry ID 0, so a consumer group always had to replay the entire
stream history before receiving new messages. Added a streamStartID
option so callers can set it to $ to only receive messages published
after the group is created, while defaulting to 0 to keep existing
deployments unaffected.

Signed-off-by: Akanksha Trehun <akankshatrehun@gmail.com>

Author: magic-peach

common/component/redis/redis.go

@@ -139,6 +139,7 @@ func ParseClientFromProperties(properties map[string]string, componentType metad
 		settings.RedeliverInterval = 15 * time.Second
 		settings.QueueDepth = 100
 		settings.Concurrency = 10
+		settings.StreamStartID = "0"
 	}
 
 	err := settings.Decode(properties)

common/component/redis/settings.go

@@ -113,6 +113,8 @@ type Settings struct {
 	QueueDepth uint `mapstructure:"queueDepth" mdonly:"pubsub"`
 	// The number of concurrent workers that are processing messages
 	Concurrency uint `mapstructure:"concurrency" mdonly:"pubsub"`
+	// The stream entry ID a new consumer group starts reading from
+	StreamStartID string `mapstructure:"streamStartID" mdonly:"pubsub"`
 
 	// The max len of stream
 	MaxLenApprox int64 `mapstructure:"maxLenApprox" mdonly:"pubsub"`

pubsub/redis/metadata.yaml

@@ -223,6 +223,15 @@ metadata:
       The number of concurrent workers that are processing messages.
     example: "15"
     default: "10"
+  - name: streamStartID
+    required: false
+    description: |
+      The stream entry ID a new consumer group starts reading from. Use "0" to
+      read the entire stream history, or "$" to only receive messages published
+      after the consumer group is created.
+    example: "$"
+    default: "0"
+    type: string
     type: number
   - name: redisType
     required: false

pubsub/redis/redis.go

@@ -39,6 +39,7 @@ const (
 	concurrency       = "concurrency"
 	maxLenApprox      = "maxLenApprox"
 	streamTTL         = "streamTTL"
+	streamStartID     = "streamStartID"
 )
 
 // redisStreams handles consuming from a Redis stream using
@@ -126,7 +127,7 @@ func (r *redisStreams) Publish(ctx context.Context, req *pubsub.PublishRequest)
 }
 
 func (r *redisStreams) CreateConsumerGroup(ctx context.Context, stream string) error {
-	err := r.client.XGroupCreateMkStream(ctx, stream, r.clientSettings.ConsumerID, "0")
+	err := r.client.XGroupCreateMkStream(ctx, stream, r.clientSettings.ConsumerID, r.clientSettings.StreamStartID)
 	// Ignore BUSYGROUP errors
 	if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
 		r.logger.Errorf("redis streams: %s", err)

pubsub/redis/redis_test.go

@@ -49,10 +49,11 @@ func TestPublishWhenClosedIsTerminal(t *testing.T) {
 
 func getFakeProperties() map[string]string {
 	return map[string]string{
-		consumerID:   "fakeConsumer",
-		enableTLS:    "true",
-		maxLenApprox: "1000",
-		streamTTL:    "1h",
+		consumerID:    "fakeConsumer",
+		enableTLS:     "true",
+		maxLenApprox:  "1000",
+		streamTTL:     "1h",
+		streamStartID: "$",
 	}
 }
 
@@ -73,6 +74,7 @@ func TestParseRedisMetadata(t *testing.T) {
 		assert.Equal(t, fakeProperties[consumerID], m.ConsumerID)
 		assert.Equal(t, int64(1000), m.MaxLenApprox)
 		assert.Equal(t, 1*time.Hour, m.StreamTTL)
+		assert.Equal(t, "$", m.StreamStartID)
 	})
 
 	// TODO: fix the code to return the error for the missing property to make this test work