Source connector duplicates regular Pub/Sub messages on a Redis cluster

[jaredpetersen]

The connector listens to all Redis cluster nodes for Pub/Sub messages. This is great for keyspace notifications, which only publish node-local messages.

However, if you use it for regular Pub/Sub, you end up with duplications:

redis-cluster:6379> publish loremipsum lorem
(integer) 1
redis-cluster:6379> publish loremipsum ipsum
(integer) 1
redis-cluster:6379> publish loremipsum dolor
(integer) 1
redis-cluster:6379> publish loremipsum si
(integer) 1
redis-cluster:6379> publish loremipsum amet
(integer) 1

[appuser@kafka-tail-records ~]$ kafka-avro-console-consumer \
>     --bootstrap-server kafka-broker-0.kafka-broker:9092 \
>     --property schema.registry.url='http://kafka-schema-registry:8081' \
>     --property print.key=true \
>     --property key.separator='|' \
>     --topic redis.events
{"channel":"loremipsum","pattern":null}|{"message":"lorem"}
{"channel":"loremipsum","pattern":null}|{"message":"lorem"}
{"channel":"loremipsum","pattern":null}|{"message":"lorem"}
{"channel":"loremipsum","pattern":null}|{"message":"ipsum"}
{"channel":"loremipsum","pattern":null}|{"message":"ipsum"}
{"channel":"loremipsum","pattern":null}|{"message":"ipsum"}
{"channel":"loremipsum","pattern":null}|{"message":"dolor"}
{"channel":"loremipsum","pattern":null}|{"message":"dolor"}
{"channel":"loremipsum","pattern":null}|{"message":"dolor"}
{"channel":"loremipsum","pattern":null}|{"message":"si"}
{"channel":"loremipsum","pattern":null}|{"message":"si"}
{"channel":"loremipsum","pattern":null}|{"message":"si"}
{"channel":"loremipsum","pattern":null}|{"message":"amet"}
{"channel":"loremipsum","pattern":null}|{"message":"amet"}
{"channel":"loremipsum","pattern":null}|{"message":"amet"}

This is NOT an issue for keyspace notifications, as those are node-local and we only subscribe to the upstream nodes and not the replicas.

While keyspace notifications are the primary reason I wrote this connector in the first place, we should also be able to support regular Pub/Sub without duplications.

I think we have a couple of ways that we can go about this:

We can add a new configuration property that indicates that you're subscribing to keyspace notifications. When enabled, we'll keep up with the cluster topology very carefully and subscribe to every node like we do now. When disabled, we can be more lax and just use the default Lettuce configuration for pub/sub. We'll still need to make sure we can handle cluster topology changes, but we won't subscribe on every node.

Alternatively, we can detect that you're subscribing to keyspace notifications based on the channel/pattern. I seriously doubt that this is feasible with patterns in play. While we know that keyspace notifications are always prefixed with __keyspace or __keyevent, I don't think I can come up with a regex fancy enough to detect people doing clever patterns like __k*y* or even *k*y*. And while these patterns could work for keyspace notifications, they could also work for other legitimate use cases if keyspace notifications aren't enabled for the Redis cluster to begin with.

I know that there's a new shard subscribe mode in Redis 7 too so we should be thinking about making this new configuration more generic so that we can support that in a future release if there's demand for it. So rather than creating a config like "keyspace_notifications_enabled" we should think about passing in a "subscribe_mode" configuration instead with string values like "all" (all nodes), "global" (global publish), "shard" (shard publish), etc.