[Bug] Redis sink: cluster mode silently ignores connectTimeout, tcpNoDelay and keepAlive

[david-streamlio]

Motivation

RedisSession.create() builds socket options — connectTimeout, tcpNoDelay, keepAlive — from
the sink config, applies them in the STANDALONE branch, and then drops them in the CLUSTER
branch. The result is that three documented configuration options are silently ignored whenever
clientMode is Cluster.

redis/src/main/java/org/apache/pulsar/io/redis/RedisSession.java:

SocketOptions socketOptions = SocketOptions.builder()
        .connectTimeout(Duration.ofMillis(config.getConnectTimeout()))
        .tcpNoDelay(config.isTcpNoDelay())
        .keepAlive(config.isKeepAlive())
        .build();

// STANDALONE: applied
ClientOptions.builder()
        .socketOptions(socketOptions)
        .build();

// CLUSTER: never applied
ClusterClientOptions.builder()
        .topologyRefreshOptions(topologyRefreshOptions)
        .build();

Nothing warns, so the only symptom is that a tuned value has no effect. A user who sets
connectTimeout: 30000 on a cluster still gets Lettuce's 10 second default.

This is pre-existing and independent of any open PR, but it is becoming more likely to bite:
TLS support for the Redis sink is being added in #126, and a TLS handshake during connection setup
is exactly the situation where connectTimeout most often needs raising. The example
configuration in the upstream request (apache/pulsar#26161) uses clientMode: "Cluster" with TLS
against AWS MemoryDB, which is precisely the combination affected.

Goal

Apply the socket options in cluster mode as well:

ClusterClientOptions.builder()
        .topologyRefreshOptions(topologyRefreshOptions)
        .socketOptions(socketOptions)
        .build();

ClusterClientOptions.Builder extends ClientOptions.Builder, so socketOptions(...) is
available and needs no other change.

Testing

RedisSessionTest already asserts against objects built by RedisSession, so a case that builds
a cluster-mode session and asserts the resulting ClusterClientOptions.getSocketOptions() carries
the configured connectTimeout would cover this without needing a live Redis.

Notes

Good first issue — the fix is one line and the surrounding test scaffolding already exists.

Two related pre-existing behaviours were noticed while looking at this and are deliberately not
in scope here; they are recorded only so they are not lost:

• STANDALONE mode silently uses only redisURIs.get(0) when several redisHosts are configured.
• withDatabase() is applied in cluster mode, where Redis rejects a non-zero SELECT.

[Ayush0612005]

I'd like to work on this issue. I'll inspect the existing RedisSessionTest coverage, add a regression test for the cluster-mode socket options, and apply the missing socket options to ClusterClientOptions.