default RedisMapper createRedisMapper(Map<String, String> properties) {
        String ttl = properties.get(REDIS_KEY_TTL);
        String additionalKey = properties.get(REDIS_ADDITIONAL_KEY);
        try {
            Class redisMapper = Class.forName(this.getClass().getCanonicalName());
            if (ttl == null && additionalKey == null) {
                return (RedisMapper) redisMapper.newInstance();
            }
            if (additionalKey != null && ttl != null) {
                return (RedisMapper) redisMapper.getConstructor(Integer.class, String.class)
                        .newInstance(ttl, additionalKey);
            }
            if (additionalKey != null) {
                return (RedisMapper) redisMapper.getConstructor(String.class)
                        .newInstance(additionalKey);
            }
            return (RedisMapper) redisMapper.getConstructor(Integer.class)
                    .newInstance(ttl);
        } catch (Exception e) {
            LOGGER.error("create redis mapper failed", e);
            throw new RuntimeException(e);
        }
    }
create a correct redis mapper use properties.

@param properties to create redis mapper.
@return redis mapper.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/common/handler/RedisMapperHandler.java