@Override
    public StreamSink getFromEntity(StreamSinkEntity entity) {
        RedisSink sink = new RedisSink();
        if (entity == null) {
            return sink;
        }

        RedisSinkDTO dto = RedisSinkDTO.getFromJson(entity.getExtParams());
        if (StringUtils.isBlank(dto.getHost())) {
            if (StringUtils.isBlank(entity.getDataNodeName())) {
                throw new BusinessException(ErrorCodeEnum.SINK_INFO_INCORRECT, "redis data node is blank");
            }
            RedisDataNodeInfo dataNodeInfo = (RedisDataNodeInfo) dataNodeHelper.getDataNodeInfo(
                    entity.getDataNodeName(), entity.getSinkType());
            CommonBeanUtils.copyProperties(dataNodeInfo, dto, true);
            String clusterMode = dataNodeInfo.getClusterMode();
            dto.setClusterMode(clusterMode);
            switch (RedisClusterMode.of(clusterMode)) {
                case CLUSTER:
                    dto.setClusterNodes(dataNodeInfo.getClusterNodes());
                    break;
                case SENTINEL:
                    dto.setMasterName(dataNodeInfo.getMasterName());
                    dto.setSentinelsInfo(dataNodeInfo.getSentinelsInfo());
                    break;
                case STANDALONE:
                    dto.setHost(dataNodeInfo.getHost());
                    dto.setPort(dataNodeInfo.getPort());
                    break;
            }
        }

        CommonBeanUtils.copyProperties(entity, sink, true);
        CommonBeanUtils.copyProperties(dto, sink, true);
        List<SinkField> sinkFields = super.getSinkFields(entity.getId());
        sink.setSinkFieldList(sinkFields);
        return sink;
    }
Redis sink operator, such as save or update redis field, etc.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/redis/RedisSinkOperator.java