public <T> T withPubSubConnection(final Function<StatefulRedisClusterPubSubConnection<K, V>, T> function) {
    try {
      return retry.executeCallable(() -> executeTimer.record(() -> function.apply(pubSubConnection)));
    } catch (final Throwable t) {
      if (t instanceof RedisException) {
        throw (RedisException) t;
      } else {
        throw new RedisException(t);
      }
    }
  }
@Test
  void testRetry() {
    when(pubSubCommands.get(anyString()))
        .thenThrow(new RedisCommandTimeoutException())
        .thenThrow(new RedisCommandTimeoutException())
        .thenReturn("value");

    assertEquals("value",
        faultTolerantPubSubConnection.withPubSubConnection(connection -> connection.sync().get("key")));

    when(pubSubCommands.get(anyString()))
        .thenThrow(new RedisCommandTimeoutException())
        .thenThrow(new RedisCommandTimeoutException())
        .thenThrow(new RedisCommandTimeoutException())
        .thenReturn("value");

    assertThrows(RedisCommandTimeoutException.class,
        () -> faultTolerantPubSubConnection.withPubSubConnection(connection -> connection.sync().get("key")));
  }