[INLONG-7404][Agent] Fix error of Redis connector

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java

@@ -64,6 +64,10 @@ public class JobProfileDto {
      * oracle source
      */
     public static final String ORACLE_SOURCE = "org.apache.inlong.agent.plugin.sources.OracleSource";
+    /**
+     * redis source
+     */
+    public static final String REDIS_SOURCE = "org.apache.inlong.agent.plugin.sources.RedisSource";
     /**
      * mqtt source
      */
@@ -215,6 +219,22 @@ private static PostgreSQLJob getPostgresJob(DataConfig dataConfigs) {
         return postgreSQLJob;
     }
 
+    private static RedisJob getRedisJob(DataConfig dataConfig) {
+        RedisJob.RedisJobConfig config = GSON.fromJson(dataConfig.getExtParams(), RedisJob.RedisJobConfig.class);
+        RedisJob redisJob = new RedisJob();
+
+        redisJob.setAuthUser(config.getUsername());
+        redisJob.setAuthPassword(config.getPassword());
+        redisJob.setHostname(config.getHostname());
+        redisJob.setPort(config.getPort());
+        redisJob.setSsl(config.getSsl());
+        redisJob.setReadTimeout(config.getTimeout());
+        redisJob.setQueueSize(config.getQueueSize());
+        redisJob.setReplId(config.getReplId());
+
+        return redisJob;
+    }
+
     private static MongoJob getMongoJob(DataConfig dataConfigs) {
 
         MongoJob.MongoJobTaskConfig config = GSON.fromJson(dataConfigs.getExtParams(),
@@ -440,6 +460,12 @@ public static TriggerProfile convertToTriggerProfile(DataConfig dataConfig) {
                 job.setSource(MONGO_SOURCE);
                 profileDto.setJob(job);
                 break;
+            case REDIS:
+                RedisJob redisJob = getRedisJob(dataConfig);
+                job.setRedisJob(redisJob);
+                job.setSource(REDIS_SOURCE);
+                profileDto.setJob(job);
+                break;
             case MQTT:
                 MqttJob mqttJob = getMqttJob(dataConfig);
                 job.setMqttJob(mqttJob);
@@ -481,6 +507,7 @@ public static class Job {
         private PostgreSQLJob postgreSQLJob;
         private OracleJob oracleJob;
         private MongoJob mongoJob;
+        private RedisJob redisJob;
         private MqttJob mqttJob;
         private SqlServerJob sqlserverJob;
     }

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/RedisJob.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.agent.pojo;
+
+import lombok.Data;
+
+@Data
+public class RedisJob {
+
+    private String authUser;
+    private String authPassword;
+    private String hostname;
+    private String port;
+    private Boolean ssl;
+    private String readTimeout;
+    private String queueSize;
+    private String replId;
+
+    @Data
+    public static class RedisJobConfig {
+
+        private String username;
+        private String password;
+        private String hostname;
+        private String port;
+        private Boolean ssl;
+        private String timeout;
+        private String queueSize;
+        private String replId;
+    }
+}

inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/RedisSource.java

@@ -40,7 +40,8 @@ public RedisSource() {
     @Override
     public List<Reader> split(JobProfile conf) {
         super.init(conf);
-        Reader redisReader = new RedisReader();
+        RedisReader redisReader = new RedisReader();
+        redisReader.setReadSource(conf.getInstanceId());
         List<Reader> readerList = new ArrayList<>();
         readerList.add(redisReader);
         sourceMetric.sourceSuccessCount.incrementAndGet();

inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/RedisReader.java

@@ -202,6 +202,10 @@ public String getReadSource() {
         return instanceId;
     }
 
+    public void setReadSource(String instanceId) {
+        this.instanceId = instanceId;
+    }
+
     @Override
     public void setReadTimeout(long mill) {
 

inlong-common/src/main/java/org/apache/inlong/common/enums/TaskTypeEnum.java

@@ -72,6 +72,8 @@ public static TaskTypeEnum getTaskType(int taskType) {
                 return MONGODB;
             case 10:
                 return TUBEMQ;
+            case 11:
+                return REDIS;
             case 12:
                 return MQTT;
             case 13: