public void eval(Object... keys) {
        RowData keyRow = GenericRowData.of(keys);
        if (cache != null) {
            RowData cachedRow = cache.getIfPresent(keyRow);
            if (cachedRow != null) {
                collect(cachedRow);
                return;
            }
        }
        for (int retry = 0; retry <= maxRetryTimes; retry++) {
            try {
                RowData rowData;
                switch (redisCommand) {
                    case GET:
                        rowData = GenericRowData
                                .of(StringData.fromString(keys[0].toString()), StringData
                                        .fromString(this.redisCommandsContainer.get(keys[0].toString())));
                        break;
                    case HGET:
                        rowData = GenericRowData
                                .of(StringData.fromString(keys[0].toString()), StringData.fromString(
                                        this.redisCommandsContainer.hget(this.additionalKey, keys[0].toString())));
                        break;
                    case ZREVRANK:
                        rowData = GenericRowData
                                .of(StringData.fromString(keys[0].toString()),
                                        this.redisCommandsContainer.zrevrank(this.additionalKey, keys[0].toString()));
                        break;
                    case ZSCORE:
                        rowData = GenericRowData
                                .of(StringData.fromString(keys[0].toString()),
                                        this.redisCommandsContainer.zscore(this.additionalKey, keys[0].toString()));
                        break;
                    default:
                        throw new UnsupportedOperationException(
                                String.format("Unsupported for redisCommand: %s", redisCommand));
                }
                if (sourceMetricData != null) {
                    sourceMetricData.outputMetricsWithEstimate(rowData);
                }
                if (cache == null) {
                    collect(rowData);
                } else {
                    collect(rowData);
                    cache.put(keyRow, rowData);
                }
                break;
            } catch (Exception e) {
                LOG.error(String.format("Redis query error, retry times = %d", retry), e);
                if (retry >= maxRetryTimes) {
                    throw new RuntimeException("Redis query error failed.", e);
                }
                try {
                    Thread.sleep(1000 * retry);
                } catch (InterruptedException e1) {
                    throw new RuntimeException(e1);
                }
            }
        }
    }
This is a lookup method which is called by Flink framework in runtime, only support one key

@param keys lookup keys
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/source/RedisRowDataLookupFunction.java