@Override
    public RFuture<V> pollAsync(long timeout, TimeUnit unit) {
        if (timeout < 0) {
            return new CompletableFutureWrapper<>((V) null);
        }
        return commandExecutor.writeAsync(getRawName(), codec, RedisCommands.BLPOP_VALUE, getRawName(), toSeconds(timeout, unit));
    }
@Test
    public void testPollWithBrokenConnection() throws InterruptedException, ExecutionException {
        GenericContainer<?> redis = createRedis();
        redis.start();

        Config config = createConfig(redis);
        RedissonClient redisson = Redisson.create(config);
        RBlockingQueue<Integer> queue1 = getQueue(redisson);
        RFuture<Integer> f = queue1.pollAsync(5, TimeUnit.SECONDS);

        Assertions.assertThrows(TimeoutException.class, () -> {
            f.toCompletableFuture().get(1, TimeUnit.SECONDS);
        });
        redis.stop();

        long start = System.currentTimeMillis();
        assertThat(f.get()).isNull();
        assertThat(System.currentTimeMillis() - start).isGreaterThan(3800);
        
        redisson.shutdown();
    }