[improve][io] Add Redis 6+ ACL username/password and TLS support to Redis sink connector

[razzopardift]

Search before reporting

• I searched in the issues and found nothing similar.

Motivation

The built-in Pulsar Redis sink connector currently exposes only a redisPassword configuration field. Internally, this maps to legacy password-only authentication (AUTH <password>), which authenticates as the default user.
This is insufficient for deployments that use Redis 6+ ACL authentication, where a username and password are required (AUTH <username> <password>).
We have run into this limitation when connecting Pulsar to ACL-enabled Redis clusters:

• self-managed Redis Cluster on Kubernetes with ACL users
• AWS — AWS MemoryDB for Redis, which enforces ACL-based authentication and requires TLS
  The current connector configuration schema (see Redis sink connector docs) only documents:
• redisHosts
• redisPassword
• redisDatabase
• connection tuning options (clientMode, timeouts, batch settings, etc.)

There is no way to configure:

1. A Redis ACL username
2. TLS/SSL for the Redis client connection

As a result, the out-of-the-box connector cannot connect to ACL + TLS Redis deployments such as AWS MemoryDB without maintaining a custom/patched NAR.

Solution

Extend the Redis sink connector to support Redis 6+ ACL authentication and optional TLS, while preserving backwards compatibility with existing password-only configurations.

1. RedisAbstractConfig.java

Add two new configuration fields:

@FieldDoc(
    required = false,
    defaultValue = "",
    sensitive = true,
    help = "The username for Redis 6+ ACL authentication")
private String redisUser;

@FieldDoc(
    required = false,
    defaultValue = "false",
    help = "Enable TLS/SSL when connecting to Redis (required for services like AWS MemoryDB)")
private boolean redisUseTls = false;

Notes:

• redisUser should be marked sensitive = true, consistent with redisPassword
• redisUseTls should default to false so existing non-TLS deployments are unaffected

2. RedisSession.java

Update redisURIs() to build RedisURI instances using Lettuce's ACL and TLS APIs:

private static List<RedisURI> redisURIs(List<HostAndPort> hostAndPorts, RedisSinkConfig config) {
    List<RedisURI> redisURIs = Lists.newArrayList();
    for (HostAndPort hostAndPort : hostAndPorts) {
        RedisURI.Builder builder = RedisURI.builder();
        builder.withHost(hostAndPort.getHost());
        builder.withPort(hostAndPort.getPort());
        builder.withDatabase(config.getRedisDatabase());
        builder.withSsl(config.isRedisUseTls());
        if (!StringUtils.isBlank(config.getRedisUser()) && !StringUtils.isBlank(config.getRedisPassword())) {
            // Redis 6+ ACL auth: username + password
            builder.withAuthentication(config.getRedisUser(), config.getRedisPassword());
        } else if (!StringUtils.isBlank(config.getRedisPassword())) {
            // Legacy auth: password only (authenticates as "default" user)
            builder.withPassword(config.getRedisPassword().toCharArray());
        }
        redisURIs.add(builder.build());
    }
    return redisURIs;
}

Behaviour:

• If both redisUser and redisPassword are set → use ACL authentication
• If only redisPassword is set → preserve existing legacy behaviour
• redisUseTls controls whether Lettuce connects with SSL/TLS

No changes are required in RedisSink.java.

3. Example sink configuration

configs:
  archive: /pulsar/connectors/pulsar-io-redis-4.0.4.nar
  tenant: bu
  namespace: cars
  name: redis-sink-example
  inputs:
    - persistent://bu/cars/example-topic
  redisHosts: "redis:6379"
  redisUser: "${REDIS_USER}"
  redisPassword: "${REDIS_PASSWORD}"
  redisDatabase: 0
  clientMode: "Cluster"
  redisUseTls: true
  operationTimeout: 3000
  batchSize: 100
  batchTimeMs: 1000

Backwards compatibility

This change should be fully backwards compatible:

Existing config	Expected behaviour
redisPassword only	Unchanged — legacy password auth
redisUser + redisPassword	New — Redis 6+ ACL auth
redisUseTls: false (default)	Unchanged — plain TCP
redisUseTls: true	New — TLS-enabled connection

Alternatives

Custom/patched NAR only — works, but forces every user of ACL/TLS Redis to maintain a forked connector build

Anything else?

Environment Tests

• Pulsar version tested against: 4.0.4
• Redis targets:
  • Redis 6+ ACL-enabled cluster (Kubernetes)
  • AWS MemoryDB for Redis (ACL + TLS required)
• Connector module: pulsar-io/redis

Are you willing to submit a PR?

• I'm willing to submit a PR!

[kritharth2005]

Hi @razzopardift , I see this hasn't had a PR linked in a while. If you aren't currently working on it, I'd love to pick this up as my first Pulsar contribution!

[razzopardift]

Hi @razzopardift , I see this hasn't had a PR linked in a while. If you aren't currently working on it, I'd love to pick this up as my first Pulsar contribution!

Hi @kritharth2005 please be my guest :-)