[improve][io] Add Redis 6+ ACL username/password and TLS support to Redis sink connector
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before reporting
- [x] I searched in the [issues](https://github.com/apache/pulsar/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 `), 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 `).
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](https://aws.amazon.com/memorydb/), which enforces ACL-based authentication and **requires TLS**
The current connector configuration schema (see [Redis sink connector docs](https://pulsar.apache.org/docs/4.0.x/io-redis-sink/)) 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 redisURIs(List hostAndPorts, RedisSinkConfig config) {
List 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?
- [x] I'm willing to submit a PR!
Contributor guide
Research direction
Start with pulsar-io/redis's RedisAbstractConfig.java and RedisSession.java, then inspect the existing Redis sink connector tests and the Redis sink documentation. Done means redisUser and redisUseTls are documented and supported, password-only authentication remains compatible, ACL authentication works when both credentials are set, and TLS is enabled when requested.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, redis
- Domain
- backend, databases
- Issue type
- Feature
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 72/100