@Test
    public void testSourceWithZrevRank() {
        StreamExecutionEnvironment executionEnv =
                StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv =
                StreamTableEnvironment.create(executionEnv);
        Jedis jedis = new Jedis("127.0.0.1", redisPort);

        jedis.zadd("rank", 10, "1");
        jedis.zadd("rank", 20, "2");
        jedis.zadd("rank", 30, "3");

        String dim = "CREATE TABLE dim (" +
                "    member_test varchar," +
                "    member_rank bigint" +
                ") WITH (" +
                "  'connector' = 'redis-inlong'," +
                "  'command' = 'zrevrank'," +
                "  'host' = 'localhost'," +
                "  'port' = '" + redisPort + "'," +
                "  'maxIdle' = '8'," +
                "  'minIdle' = '1'," +
                "  'maxTotal' = '2'," +
                "  'timeout' = '2000'," +
                "  'additional.key' = 'rank'" +
                ")";

        String source =
                "create table source(member_test varchar, proctime as procTime()) "
                        + "with ('connector'='datagen',  'rows-per-second'='1', "
                        + "'fields.member_test.kind'='sequence',  'fields.member_test.start'='1',  'fields.member_test.end'='3'"
                        + ")";
        String sink = "CREATE TABLE sink (" +
                "    aaa STRING," +
                "    bbb STRING," +
                "    ccc BIGINT," +
                "    PRIMARY KEY (`aaa`) NOT ENFORCED" +
                ") WITH (" +
                "  'connector' = 'redis-inlong'," +
                "  'sink.batch-size' = '1'," +
                "  'format' = 'csv'," +
                "  'data-type' = 'HASH'," +
                "  'redis-mode' = 'standalone'," +
                "  'host' = '127.0.0.1'," +
                "  'port' = '" + redisPort + "'," +
                "  'maxIdle' = '8'," +
                "  'minIdle' = '1'," +
                "  'maxTotal' = '2'," +
                "  'timeout' = '2000'" +
                ")";

        tableEnv.executeSql(dim);
        tableEnv.executeSql(source);
        tableEnv.executeSql(sink);
        String sql =
                " insert into sink"
                        + " select 'rank_test',s.member_test, d.member_rank from source s"
                        + "  left join dim for system_time as of s.proctime as d "
                        + "on d.member_test = s.member_test";

        tableEnv.executeSql(sql);

        await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
            assertEquals("2", jedis.hget("rank_test", "1"));
            assertEquals("1", jedis.hget("rank_test", "2"));
            assertEquals("0", jedis.hget("rank_test", "3"));
        });
    }
hget only support get data from the given one key
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/redis/src/test/java/org/apache/inlong/sort/redis/RedisTableTest.java