@Override
    public DynamicTableSink createDynamicTableSink(Context context) {
        Map<String, String> properties = context.getCatalogTable().getOptions();
        FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
        ReadableConfig config = helper.getOptions();
        RedisDataType dataType = config.get(DATA_TYPE);
        SchemaMappingMode schemaMappingMode = config.get(SCHEMA_MAPPING_MODE);

        validateDataTypeAndSchemaMappingMode(dataType, schemaMappingMode);
        String inlongMetric = config.getOptional(INLONG_METRIC).orElse(INLONG_METRIC.defaultValue());
        String auditHostAndPorts = config.getOptional(INLONG_AUDIT).orElse(INLONG_AUDIT.defaultValue());

        ResolvedSchema resolvedSchema = context.getCatalogTable().getResolvedSchema();
        EncodingFormat<SerializationSchema<RowData>> encodingFormat = helper
                .discoverEncodingFormat(SerializationFormatFactory.class, FactoryUtil.FORMAT);

        return new RedisDynamicTableSink(
                encodingFormat,
                resolvedSchema,
                dataType,
                schemaMappingMode,
                config,
                properties,
                inlongMetric,
                auditHostAndPorts);
    }
Supported redis source commands, contain [GET|HGET|ZREVRANK|ZSCORE] at now.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/table/RedisDynamicTableFactory.java