`UnifiedJedis.subscribe()` returns a still-subscribed connection to the pool when the subscribe loop exits abnormally

Title: UnifiedJedis.subscribe() returns a still-subscribed connection to the pool when the subscribe loop exits abnormally

Expected behavior

When the subscribe loop in JedisPubSubBase.process() terminates while the connection is still subscribed, the connection should be marked broken so that Connection.close() routes it to ConnectionPool.returnBrokenResource(). A connection that is still in subscriber mode on the server must never be handed to the next borrower.

Actual behavior

UnifiedJedis.subscribe() borrows a pooled connection in a try-with-resources:

// UnifiedJedis.java:3819
public void subscribe(final JedisPubSub jedisPubSub, final String... channels) {
  try (Connection connection = this.provider.getConnection()) {
    jedisPubSub.proceed(connection, channels);
  }
}

proceed() only rolls back the timeout on the way out:

// JedisPubSubBase.java:87
public final void proceed(Connection client, T... channels) {
  this.client = client;
  this.client.setTimeoutInfinite();
  try {
    subscribe(channels);
    process();
  } finally {
    this.client.rollbackTimeout();   // <- broken is never set
  }
}

and the loop has two exits that leave the subscription live:

// JedisPubSubBase.java:178
} while (!Thread.currentThread().isInterrupted() && isSubscribed());

1. A callback throws. Anything escaping onMessage / onPMessage / onSubscribe / onUnsubscribe propagates out of process(). rollbackTimeout() succeeds (the socket is still open), so broken stays false.
2. The listener thread's interrupt flag is set. The loop exits after the current message, still subscribed, and subscribe() returns normally. Because the interrupt check is evaluated before isSubscribed(), user code cannot even send an UNSUBSCRIBE and let the loop drain the confirmations — the loop will not run another iteration.
3. process() itself throws JedisException("Unknown message type: ...") (lines 121, 166, 176), or a JedisDataException comes off the read path.

In all three cases Connection.broken remains false — readProtocolWithCheckingBroken() (line 386) only sets it on JedisConnectionException — so Connection.close() (line 255) takes the returnResource() branch and a connection that is still SUBSCRIBEd goes back into the shared pool.

It does not self-heal. The next borrower issuing e.g. GET receives -ERR ... only (P|S)SUBSCRIBE / (P|S)UNSUBSCRIBE / PING / QUIT / RESET are allowed in this context. That is a JedisDataException, which does not set broken either, so the connection is returned to the pool "clean" again and the next borrower hits the same error. It can bounce indefinitely.

Worse, it can corrupt replies silently. If anything is still publishing to the subscribed channels, the socket has queued push messages. Protocol.process() reads a RESP3 push (>) as a plain multi-bulk (Protocol.java:154), and the base Connection.protocolReadPushes() is an empty stub (line 383, only overridden by CacheConnection). So replies are shifted by one, and a command that legitimately expects a list (MGET, HGETALL, SMEMBERS, LRANGE) can return the pubsub payload as data instead of throwing. RESP3 does not help here.

The default pool configuration does not catch it: ConnectionPoolConfig sets only testWhileIdle(true) with a 30s evictor and a 60s minEvictableIdleTime, and testOnBorrow / testOnReturn default to false, so on a busy pool the connection is handed out long before the evictor would test it.

A secondary problem: the this.client = null invalidation at the end of process() is commented out (JedisPubSubBase.java:180-181), so after proceed() returns the JedisPubSub still references a connection that now belongs to a different borrower. A late unsubscribe() from a shutdown thread writes UNSUBSCRIBE into someone else's in-flight connection.

Steps to reproduce:

Please create a reproducible case of your problem. Make sure that case repeats consistently and it's not random

1. Start a Redis server on localhost:6379.
2. Run the program below. maxTotal(1) guarantees that the next borrow reuses the connection the subscriber abandoned.
3. The final GET fails with JedisDataException: ERR Can't execute 'get': only (P|S)SUBSCRIBE / (P|S)UNSUBSCRIBE / PING / QUIT / RESET are allowed in this context. It keeps failing on subsequent borrows too — the JedisDataException does not set broken, so the connection is returned to the pool each time; only the testWhileIdle evictor can eventually clear it, and that requires the connection to sit idle for minEvictableIdleTime (60s by default).

import java.util.concurrent.CountDownLatch;

import redis.clients.jedis.ConnectionPoolConfig;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisPooled;
import redis.clients.jedis.JedisPubSub;

public class SubscribedConnectionReturnedToPool {

    public static void main(String[] args) throws Exception {
        HostAndPort node = new HostAndPort("localhost", 6379);

        ConnectionPoolConfig poolConfig = new ConnectionPoolConfig();
        poolConfig.setMaxTotal(1); // the next borrow is necessarily the abandoned connection

        try (JedisPooled subscriberClient = new JedisPooled(node, poolConfig);
             JedisPooled publisherClient = new JedisPooled(node)) {

            CountDownLatch subscribed = new CountDownLatch(1);

            JedisPubSub pubSub = new JedisPubSub() {
                @Override
                public void onSubscribe(String channel, int subscribedChannels) {
                    subscribed.countDown();
                }

                @Override
                public void onMessage(String channel, String message) {
                    // case 1 - a listener that throws
                    throw new RuntimeException("listener failure");

                    // case 2 - an interrupted listener thread. Comment out the throw above and
                    // uncomment the line below: subscribe() then returns *normally* and the
                    // connection is still returned to the pool while subscribed.
                    // Thread.currentThread().interrupt();
                }
            };

            Thread subscriber = new Thread(() -> {
                try {
                    subscriberClient.subscribe(pubSub, "some-channel");
                    System.out.println("subscribe() returned normally");
                } catch (RuntimeException e) {
                    System.out.println("subscribe() threw " + e);
                }
            });
            subscriber.start();
            subscribed.await();

            publisherClient.publish("some-channel", "hello");
            subscriber.join();

            // the abandoned connection is back in the pool with broken == false
            System.out.println("GET -> " + subscriberClient.get("any-key"));
        }
    }
}

Redis / Jedis Configuration

Jedis version:

5.2.0. The same code is present in 6.0.0 — UnifiedJedis.subscribe() is unchanged (6.0.0 line 3870) and the this.client = null invalidation is still commented out (line 199).

Redis version:

Not server-version specific: it follows from the RESP2 subscriber-mode command restrictions, and the reply-shifting variant affects RESP3 as well.

Java version:

21

Suggested fix

Connection.setBroken() is already public (Connection.java:295), so proceed() / proceedWithPatterns() can do this themselves:

public final void proceed(Connection client, T... channels) {
  this.client = client;
  this.client.setTimeoutInfinite();
  try {
    subscribe(channels);
    process();
  } catch (Throwable t) {
    this.client.setBroken();
    throw t;
  } finally {
    if (isSubscribed()) {
      // the loop gave up (interrupted, most likely) while the server still has us subscribed
      this.client.setBroken();
    }
    this.client.rollbackTimeout();
    this.client = null;
  }
}

Notes on that patch:

• setBroken() does not close the socket, so the subsequent rollbackTimeout() still succeeds and the original exception is not replaced.
• Setting the flag is conveniently self-terminating for the interrupt case: getUnflushedObject() delegates straight to readProtocolWithCheckingBroken() (Connection.java:348), which throws as soon as broken is set, so the loop cannot continue on a connection that has been given up on.
• Restoring this.client = null (currently commented out) is a behaviour change worth calling out explicitly: an unsubscribe() issued after proceed() has returned would then get the intended JedisException("... is not connected to a Connection.") from sendAndFlushCommand() (line 39) instead of silently writing into a recycled connection.

Two related points you may want to consider separately:

1. The loop condition evaluates !Thread.currentThread().isInterrupted() before isSubscribed(), which means an interrupted subscriber can never drain an UNSUBSCRIBE — there is no clean shutdown path once the flag is set. Checking isSubscribed() first would at least let a pending unsubscribe complete. Related: Thread.interrupt() does not unblock a plain socket read, so interrupting a subscriber thread is not a working shutdown mechanism at all today (ExecutorService.shutdownNow() will not stop a Jedis subscriber).
2. A subscriber connection is monopolised for its entire life, so borrowing it from the shared pool has no upside and this failure mode as the downside. UnifiedJedis.subscribe() could reasonably use a dedicated connection instead.

Workaround for users

Do not use UnifiedJedis.subscribe(); own the connection at the call site, where ConnectionProvider.getConnection() and Connection.setBroken() are both public:

Connection conn = provider.getConnection();
try {
    pubSub.proceed(conn, channels);
} catch (Throwable t) {
    conn.setBroken();
    throw t;
} finally {
    if (pubSub.isSubscribed()) {
        conn.setBroken();
    }
    conn.close();
}

Setting testOnBorrow(true) also limits the blast radius — ConnectionFactory.validateObject() is isConnected() && ping(), and in subscriber mode RESP2 PING replies with a 2-element array ["pong", ""] which fails the byte[] cast in getStatusCodeReply(), so the connection gets destroyed instead of reused.