@Override
    public Flux<BooleanResponse<RenameCommand>> rename(Publisher<RenameCommand> commands) {

        return execute(commands, command -> {
            Assert.notNull(command.getKey(), "Key must not be null!");
            Assert.notNull(command.getNewKey(), "New name must not be null!");

            byte[] keyBuf = toByteArray(command.getKey());
            byte[] newKeyBuf = toByteArray(command.getNewKey());

            if (executorService.getConnectionManager().calcSlot(keyBuf) == executorService.getConnectionManager().calcSlot(newKeyBuf)) {
                return super.rename(commands);
            }

            return read(keyBuf, ByteArrayCodec.INSTANCE, RedisCommands.DUMP, keyBuf)
                    .filter(Objects::nonNull)
                    .zipWith(
                            Mono.defer(() -> pTtl(command.getKey())
                                    .filter(Objects::nonNull)
                                    .map(ttl -> Math.max(0, ttl))
                                    .switchIfEmpty(Mono.just(0L))
                            )
                    )
                    .flatMap(valueAndTtl -> {
                        return write(newKeyBuf, StringCodec.INSTANCE, RedisCommands.RESTORE, newKeyBuf, valueAndTtl.getT2(), valueAndTtl.getT1());
                    })
                    .thenReturn(new BooleanResponse<>(command, true))
                    .doOnSuccess((ignored) -> del(command.getKey()));
        });
    }
@Test
    public void testRename_keyNotExist() {
        Integer originalSlot = getSlotForKey(originalKey);
        newKey = getNewKeyForSlot(new String(originalKey.array()), getTargetSlot(originalSlot));

        if (sameSlot) {
            // This is a quirk of the implementation - since same-slot renames use the non-cluster version,
            // the result is a Redis error. This behavior matches other spring-data-redis implementations
            assertThatThrownBy(() -> connection.keyCommands().rename(originalKey, newKey).block())
                    .isInstanceOf(RedisSystemException.class);

        } else {
            Boolean response = connection.keyCommands().rename(originalKey, newKey).block();

            assertThat(response).isTrue();

            final ByteBuffer newKeyValue = connection.stringCommands().get(newKey).block();
            assertThat(newKeyValue).isEqualTo(null);
        }
    }