@Override
    public Map<String, String> tableOptions() {
        Map<String, String> options = super.tableOptions();
        options.put(RedisConstant.CONNECTOR_KEY, RedisConstant.CONNECTOR);
        options.put(RedisConstant.COMMAND, command.getValue());
        options.put(RedisConstant.REDIS_MODE, redisMode.getValue());
        switch (redisMode) {
            case CLUSTER:
                options.put(RedisConstant.CLUSTER_NODES,
                        Preconditions.checkNotNull(clusterNodes, "clusterNodes is null"));
                break;
            case SENTINEL:
                options.put(RedisConstant.MASTER_NAME, Preconditions.checkNotNull(masterName,
                        "masterName is null"));
                options.put(RedisConstant.SENTINELS_INFO,
                        Preconditions.checkNotNull(sentinelsInfo, "sentinelsInfo is null"));
                if (database != null) {
                    options.put(RedisConstant.DATABASE, database.toString());
                }
                break;
            case STANDALONE:
                options.put(RedisConstant.HOST, Preconditions.checkNotNull(host, "host is null"));
                if (port != null) {
                    options.put(RedisConstant.PORT, port.toString());
                }
                if (database != null) {
                    options.put(RedisConstant.DATABASE, database.toString());
                }
                break;
            default:
        }
        if (command == RedisCommand.HGET || command == RedisCommand.ZREVRANK || command == RedisCommand.ZSCORE) {
            options.put(RedisConstant.ADDITIONAL_KEY,
                    Preconditions.checkNotNull(additionalKey, "additionalKey is null"));
        }
        if (password != null) {
            options.put(RedisConstant.PASSWORD, password);
        }
        if (timeout != null) {
            options.put(RedisConstant.TIMEOUT, timeout.toString());
        }
        if (soTimeout != null) {
            options.put(RedisConstant.SOTIMEOUT, soTimeout.toString());
        }
        if (maxTotal != null) {
            options.put(RedisConstant.MAXTOTAL, maxTotal.toString());
        }
        if (maxIdle != null) {
            options.put(RedisConstant.MAXIDLE, maxIdle.toString());
        }
        if (minIdle != null) {
            options.put(RedisConstant.MINIDLE, minIdle.toString());
        }
        if (lookupOptions != null) {
            if (lookupOptions.getLookupCacheMaxRows() != null) {
                options.put(RedisConstant.LOOKUP_CACHE_MAX_ROWS, lookupOptions.getLookupCacheMaxRows().toString());
            }
            if (lookupOptions.getLookupCacheTtl() != null) {
                options.put(RedisConstant.LOOKUP_CACHE_TTL, lookupOptions.getLookupCacheTtl().toString());
            }
            if (lookupOptions.getLookupMaxRetries() != null) {
                options.put(RedisConstant.LOOKUP_MAX_RETRIES, lookupOptions.getLookupMaxRetries().toString());
            }
        }
        return options;
    }
RedisExtract base constructor

@param id The id of extract node
@param name The name of extract node
@param fields The fields of extract node
@param watermarkField The watermark field of extract node
@param properties The custom properties of extract node
@param primaryKey The primary key of extract node
@param redisMode The redis deploy mode connect to redis
@param command The redis command connect to redis
@param clusterNodes The cluster node infos connect to redis only used for {@link RedisMode#CLUSTER}
@param masterName The master name connect to redis only used for {@link RedisMode#SENTINEL}
@param sentinelsInfo The sentinels connect to redis info used for {@link RedisMode#SENTINEL}
@param host The host connect to redis only used for {@link RedisMode#STANDALONE}
@param port The port connect to redis only used for {@link RedisMode#STANDALONE}
@param password The password connect to redis
@param additionalKey The additional key connect to redis only used for [Hash|Sorted-Set] data type
@param database The database connect to redis
        used for {@link RedisMode#STANDALONE} and {@link RedisMode#SENTINEL}
@param timeout The timeout connect to redis
@param soTimeout The soTimeout connect to redis
@param maxTotal The maxTotal connect to redis
@param maxIdle The maxIdle connect to redis
@param minIdle The minIdle connect to redis
@param lookupOptions The lookup options for connector redis
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/RedisExtractNode.java