apache / apache/pulsar

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

Open
#26,161 2 comments 5 reactions 0 assignees View on GitHub
type/enhancement
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.