@Override
    protected void setTargetEntity(SinkRequest request, StreamSinkEntity targetEntity) {

        if (!this.getSinkType().equals(request.getSinkType())) {
            throw new BusinessException(ErrorCodeEnum.SINK_SAVE_FAILED,
                    SINK_TYPE_NOT_SUPPORT.getMessage() + ": " + getSinkType());
        }

        RedisSinkRequest sinkRequest = (RedisSinkRequest) request;

        if (StringUtils.isNotBlank(targetEntity.getDataNodeName())) {
            RedisDataNodeInfo dataNodeInfo = (RedisDataNodeInfo) dataNodeHelper.getDataNodeInfo(
                    targetEntity.getDataNodeName(), targetEntity.getSinkType());

            CommonBeanUtils.copyProperties(dataNodeInfo, sinkRequest, true);
        }

        RedisClusterMode redisClusterMode = RedisClusterMode.of(sinkRequest.getClusterMode());

        expectNotNull(redisClusterMode,
                "Redis ClusterMode must in one of " + Arrays.toString(RedisClusterMode.values()) + " !");

        switch (redisClusterMode) {
            case CLUSTER:
                checkClusterNodes(sinkRequest.getClusterNodes());
                break;
            case SENTINEL:
                expectNotEmpty(sinkRequest.getMasterName(), "Redis MasterName of Sentinel cluster must not null!");
                expectNotEmpty(sinkRequest.getSentinelsInfo(),
                        "Redis sentinelsInfo of Sentinel cluster must not null!");
                break;
            case STANDALONE:
                String host = sinkRequest.getHost();
                Integer port = sinkRequest.getPort();

                expectNotEmpty(host, "Redis server host must not null!");
                expectTrue(
                        port != null && port > 1 && port < PORT_MAX_VALUE,
                        "The port of the redis server must be greater than 0 and less than 65535!");
                break;
        }
        RedisDataType dataType = RedisDataType.valueOf(sinkRequest.getDataType());
        expectNotNull(dataType, "Redis DataType must not null");

        RedisSchemaMapMode mapMode = RedisSchemaMapMode.valueOf(sinkRequest.getSchemaMapMode());
        expectTrue(dataType.getMapModes().contains(mapMode),
                "Redis schemaMapMode '" + mapMode + "' is not supported in '" + dataType + "'");

        try {
            RedisSinkDTO dto = RedisSinkDTO.getFromRequest(sinkRequest, targetEntity.getExtParams());
            targetEntity.setExtParams(objectMapper.writeValueAsString(dto));
        } catch (Exception e) {
            throw new BusinessException(SINK_SAVE_FAILED,
                    String.format("serialize extParams of Redis SinkDTO failure: %s", e.getMessage()));
        }
    }
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