Add TLS truststore support to Redis sink

[kritharth2005]

Motivation

Adds TLS trust configuration to the Redis sink, allowing Redis deployments
using internally signed certificates to be configured without requiring a
custom NAR.

Changes

• Add configurable TLS peer verification mode (FULL, CA, NONE).
• Add custom TLS truststore path and password configuration.
• Apply truststore SSL options to standalone and cluster Redis clients.
• Load the truststore password through Pulsar's secret mechanism.
• Add unit tests covering TLS verification, truststore configuration, and secret-based password loading.

Testing

• ./gradlew :redis:test --rerun-tasks
• ./gradlew :redis:check --rerun-tasks

Both pass successfully.

Closes #132

[david-streamlio]

Reviewed the TLS truststore wiring against Lettuce 6.5.1 and this repo's existing conventions. This is solid, well-tested work — the wiring is correct, FULL as the default means existing deployments are unaffected, and a few things here are better than typical:

• sensitive = true on the password and a getSecret() test. That is the part people forget.
• The wrong-password negative check in buildSslOptionsWithTruststorePathAndPasswordTest proves the password actually reaches Lettuce rather than being silently dropped. Most contributions assert only the happy path.
• The comment explaining why getTruststore() cannot be asserted and createSslContextBuilder() is used instead is accurate for 6.5.x — someone clearly read the library rather than guessing.
• tls/README.md documents the fixture's provenance and notes the private key was discarded. That is exactly the right way to check in a test keystore.

I also checked two things I was ready to flag and won't: redisTlsVerifyPeer as a String rather than an enum matches the existing clientMode, and getResource().getFile() matches 20 other uses in this repo including the pre-existing helper in RedisSinkConfigTest. Both are house style.

Nothing below blocks merge. The comments are ordered roughly by how much I care:

1. Invalid verify-peer values pass validate() and fail later, with a message that omits the offending value.
2. The truststore path is never checked for existence, so a typo becomes a connection-time IOException.
3. The help text advertises "JKS or PKCS12" while the store type is not configurable.
4. Disabling peer verification leaves no trace in the logs.

Plus two minor ones (locale on toUpperCase, and a pre-existing cluster-mode socketOptions gap you happen to be next to).

[david-streamlio]

validate() only null-checks this, so an invalid value still passes configuration validation and fails later, when RedisSession.create parses it. A typo ("Full ", "true", or an empty string) therefore surfaces as a connection-time failure rather than a config error.

Parsing the enum here instead would fail fast, and would let you drop the null check — SslVerifyMode.valueOf(null) throws anyway:

try {
    SslVerifyMode.valueOf(redisTlsVerifyPeer.toUpperCase(Locale.ROOT));
} catch (IllegalArgumentException | NullPointerException e) {
    throw new IllegalArgumentException("redisTlsVerifyPeer must be one of "
        + Arrays.asList(SslVerifyMode.values()) + ", got: " + redisTlsVerifyPeer);
}

[david-streamlio]

The message lists the valid values but not the one that was actually supplied, which is the single most useful thing for someone debugging their config. It also drops e as the cause.

throw new IllegalArgumentException("Illegal Redis TLS verify-peer mode '" + value
    + "', valid values are: " + Arrays.asList(SslVerifyMode.values()), e);

If you move validation into RedisAbstractConfig.validate() (see my other comment), this becomes unreachable for user input and can just carry the value for safety.

[david-streamlio]

Nothing checks that this path exists or is readable, so a wrong path surfaces as an IOException from inside Lettuce/Netty when the SSL context is built at connection time — well away from the setting that caused it.

Worth an exists() / canRead() check that names redisTlsTrustStorePath and its value, ideally in validate() so it fails before any connection is attempted. loadFromMapWithTlsTrustFieldsTest shows the gap: it loads /tmp/fake-truststore.jks and validates fine.

[david-streamlio]

The help advertises "(e.g. JKS or PKCS12)", but the store type is never configurable — SslOptions.Builder.keyStoreType(String) exists in Lettuce 6.5.1 and isn't surfaced, so the type is whatever the JVM defaults to.

The fixture added in this PR is the trap in miniature: per tls/README.md it is PKCS12 content in a file named .jks. It happens to load because the JDK's PKCS12 keystore reads JKS too under keystore.type.compat, but a user who disables that, or who has a store the default type can't read, has no escape hatch.

Either add a redisTlsTrustStoreType (blank = JVM default), or narrow this text to say only the JVM's default store type is supported.

[david-streamlio]

This is the line that turns "peer verification is always on" — which is what the old redisUseTls help text promised — into something an operator can switch off. NONE disables both chain and hostname verification, which makes the connection trivially MITM-able.

The @FieldDoc says "should not be used in production", but nothing in the running system ever says so: a cluster silently accepting any certificate looks identical to a correctly secured one in the logs. A WARN here when the mode is NONE (arguably also CA, which skips the hostname check) would make the weakened posture visible where someone would actually notice it.

[david-streamlio]

toUpperCase() uses the default locale. None of FULL/CA/NONE contain an i, so this is latent rather than a live bug, but Locale.ROOT is free and is what the rest of the ecosystem expects. (Same applies to line 93's getClientMode().toUpperCase(), which is pre-existing.)

[david-streamlio]

Not this PR's doing, but you are editing this builder: the standalone branch above sets .socketOptions(socketOptions) and this one never has, so tcpNoDelay, keepAlive and connectTimeout are silently ignored in cluster mode. Adding .socketOptions(socketOptions) here would be a welcome drive-by if you are up for it — equally fine to leave for a separate PR.

[kritharth2005]

Thanks for the detailed review! Noted — I'll work through these suggestions and push the updates once I get to them.

[kritharth2005]

Pushed updates addressing all six points from the review — replied inline
on each thread with what specifically changed. Summary:

• redisTlsVerifyPeer is now validated (and fails fast with a clear
  message listing valid values and what was received) in
  RedisAbstractConfig.validate(), instead of only surfacing at
  connection time.
• The parseVerifyMode error message in RedisSession now includes the
  offending value and the original cause — mostly defensive now that
  validate() catches invalid values first.
• redisTlsTrustStorePath is checked for existence/readability in
  validate() when TLS is enabled (gated on redisUseTls so an
  unrelated/stale path doesn't fail validation for non-TLS configs).
• Added redisTlsTrustStoreType (blank = JVM default), wired through
  Lettuce's SslOptions.Builder.keyStoreType(...), and narrowed the
  truststore-path field's help text to point at it instead of naming
  formats directly.
• Non-FULL verify modes now log a WARN, once per session rather than
  once per host.
• Fixed the locale-dependent toUpperCase() calls — both the new one and
  the pre-existing clientMode one.
• Added .socketOptions(socketOptions) to the cluster-mode branch,
  matching standalone (the drive-by you flagged — thanks for calling it
  out, took you up on it).

Also added RedisSinkTlsIntegrationTest, using Testcontainers +
redis:7-alpine (following the same pattern already used in
aerospike/hbase/nsq/solr/influxdb/mqtt), since everything else here only
exercised SslOptions construction in isolation and never actually
performed a live TLS handshake through the sink:

• A positive case: a real write through RedisSink.open()/write()/
  close() over a verified TLS connection, confirmed by reading the
  value back from Redis afterward, not just checking for no exception.
• A negative case: the same server, but a truststore that doesn't trust
  its certificate. This asserts down the actual cause chain
  (SSLHandshakeException / PKIX path-building failure) rather than
  just the outer RedisConnectionException, so the test can't
  accidentally pass for an unrelated connection failure — it specifically
  proves verification isn't silently bypassed.

One more thing I found while building that integration test, flagging
rather than fixing here: if redisTlsTrustStorePath is set for a
password-protected store but redisTlsTrustStorePassword is left blank
or wrong, buildSslOptions() currently builds without error, and the
failure only surfaces much later, deep inside the actual TLS handshake
(InvalidAlgorithmParameterException: trustAnchors must be non-empty) —
a JDK/Lettuce quirk where an unopenable protected keystore silently
yields zero trust anchors rather than failing at load time. It's the
same class of "fails far from the cause" problem the path-existence
check above addresses, just for the password instead of the path. I
didn't fix it here since it's outside what was originally flagged and
the fix is more invasive — validate() would need to actually attempt
to open the store with the configured password, not just confirm it
exists. Happy to take it on as a follow-up in this PR or a separate one
if it's worth doing — flagging in case anyone's already looking at
something adjacent, so we don't duplicate effort.

All existing tests plus the new ones pass locally
(./gradlew :redis:check --rerun-tasks).

[david-streamlio]

Re-reviewed at c9f361e7, scoped to the commit added since my last pass. CI is now green on that head — all four jobs, including the Other shard that carries the redis module.

All six points from my earlier review are addressed. I checked each against the code rather than taking the commit message for it: redisTlsVerifyPeer is validated in validate() and the message now carries the offending value; the truststore path is checked for existence and readability; redisTlsTrustStoreType is configurable and the help text points at it instead of advertising formats that were not selectable; a warning fires once when the mode is not FULL; Locale.ROOT at both toUpperCase sites; and the cluster-mode socketOptions gap I had explicitly called out of scope is fixed as well.

The integration test is the substantive addition here, and it earns its place. The previous round asserted SslOptions in isolation, which proves the builder was populated but not that a handshake works. RedisSinkTlsIntegrationTest now runs a real redis-server with TLS via Testcontainers and connects through the connector's own configuration path. That is a materially stronger claim, and the green Other shard is what makes it credible rather than a local-only result — it confirms the fixtures copy into the container correctly, the Ready to accept connections tls wait strategy works against redis:7-alpine, and the handshake succeeds end to end on a machine that is not yours.

One call that looks wrong and is not, worth a comment purely so the next reader does not stop on it as I did. buildSslOptions sets a truststore type through builder.keyStoreType(...). I went to the Lettuce source expecting a bug: SslOptions carries a single keyStoreType field and passes it into the truststore lambda (trustmanager.accept(sslContextBuilder, this.keyStoreType) at line 642), and there is no truststoreType() in 6.5.1. The call is correct, and it is the only way to set this.

Three notes, none blocking, in the order I care about them.

1. The io_uring disable is unconditional, and CI cannot tell us whether it was needed. redis/build.gradle.kts now sets io.lettuce.core.iouring=false for every test in the module, attributed in the comment to sandboxes with a low memlock ulimit. The green run shows the property is harmless on GitHub's runners — but it does not show that the tests would have failed without it there, because they never ran without it. So the repo now permanently avoids Netty's io_uring transport in redis tests to accommodate one environment, and nothing will ever re-test that assumption.

If the flake was specific to your sandbox, gating it (an env var, or a providers.environmentVariable(...) check) would keep CI exercising the default transport. If you would rather keep it unconditional, that is defensible — it is test-only and documented — but it is worth being a deliberate choice rather than a leftover.

2. The private key is now committed, and I want to be explicit that I think that is fine, because my earlier review singled out its absence as the right way to check in a test keystore, and a reader comparing the two comments would otherwise find them contradictory. tls/README.md documents the provenance, why CN=localhost is required for the Testcontainers mapped port, why this fixture is deliberately separate from redis-test-truststore.jks, and the chmod 644 requirement with the failure mode it prevents. It is a throwaway self-signed key for a local container with no trust path anywhere. The license check passing also confirms .ratignore's existing **/*.key and **/*.crt entries cover it, so no exclusion work is needed.

3. @BeforeMethod starts a container per test method — two container starts for two tests, each waiting on the log message. @BeforeClass would halve that with no loss of isolation, since neither test mutates server state the other observes.

Nothing above needs to change before merge as far as I am concerned. Thanks for the thorough follow-up on the first round — the negative-path coverage and the fixture documentation are both better than the norm here.