[Fix] [admin] Fix job monitoring bugs caused by multi tenancy (#1158)

Co-authored-by: wenmo <32723967+wenmo@users.noreply.github.com>

dlink-admin/src/main/java/com/dlink/configure/MybatisPlusConfig.java

@@ -19,7 +19,7 @@
 
 package com.dlink.configure;
 
-import com.dlink.context.RequestContext;
+import com.dlink.context.TenantContextHolder;
 
 import java.util.List;
 
@@ -46,34 +46,21 @@ import net.sf.jsqlparser.expression.NullValue;
 public class MybatisPlusConfig {
 
     private static final List<String> IGNORE_TABLE_NAMES = Lists.newArrayList(
-            "dlink_namespace"
-            , "dlink_alert_group"
-            , "dlink_alert_history"
-            , "dlink_alert_instance"
-            , "dlink_catalogue"
-            , "dlink_cluster"
-            , "dlink_cluster_configuration"
-            , "dlink_database"
-            //,"dlink_fragment"
-            , "dlink_history"
-            , "dlink_jar"
-            , "dlink_job_history"
-            , "dlink_job_instance"
-            ,"dlink_role"
-            , "dlink_savepoints"
-            , "dlink_task"
-            , "dlink_task_statement"
-            , "dlink_task_version"
-    );
+            "dlink_namespace", "dlink_alert_group", "dlink_alert_history", "dlink_alert_instance", "dlink_catalogue",
+            "dlink_cluster", "dlink_cluster_configuration", "dlink_database"
+            // ,"dlink_fragment"
+            , "dlink_history", "dlink_jar", "dlink_job_history", "dlink_job_instance", "dlink_role", "dlink_savepoints",
+            "dlink_task", "dlink_task_statement", "dlink_task_version");
 
     @Bean
     public MybatisPlusInterceptor mybatisPlusInterceptor() {
         log.info("mybatis plus interceptor execute");
         MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
         interceptor.addInnerInterceptor(new TenantLineInnerInterceptor(new TenantLineHandler() {
+
             @Override
             public Expression getTenantId() {
-                Integer tenantId = (Integer) RequestContext.get();
+                Integer tenantId = (Integer) TenantContextHolder.get();
                 if (tenantId == null) {
                     log.warn("request context tenant id is null");
                     return new NullValue();

dlink-admin/src/main/java/com/dlink/context/RequestContext.java → dlink-admin/src/main/java/com/dlink/context/TenantContextHolder.java

@@ -20,20 +20,21 @@
 package com.dlink.context;
 
 /**
- * request context
+ * TenantContextHolder
  */
-public class RequestContext {
-    private static final ThreadLocal<Object> threadLocal = new ThreadLocal<>();
+public class TenantContextHolder {
+
+    private static final ThreadLocal<Object> TENANT_CONTEXT = new ThreadLocal<>();
 
     public static void set(Object value) {
-        threadLocal.set(value);
+        TENANT_CONTEXT.set(value);
     }
 
     public static Object get() {
-        return threadLocal.get();
+        return TENANT_CONTEXT.get();
     }
 
-    public static void remove() {
-        threadLocal.remove();
+    public static void clear() {
+        TENANT_CONTEXT.remove();
     }
 }

dlink-admin/src/main/java/com/dlink/init/SystemInit.java

@@ -20,7 +20,6 @@
 package com.dlink.init;
 
 import com.dlink.assertion.Asserts;
-import com.dlink.context.RequestContext;
 import com.dlink.daemon.task.DaemonFactory;
 import com.dlink.daemon.task.DaemonTaskConfig;
 import com.dlink.job.FlinkJobTask;
@@ -76,7 +75,6 @@ public class SystemInit implements ApplicationRunner {
         List<Tenant> tenants = tenantService.list();
         sysConfigService.initSysConfig();
         for (Tenant tenant : tenants) {
-            RequestContext.set(tenant.getId());
             taskService.initDefaultFlinkSQLEnv(tenant.getId());
         }
         initTaskMonitor();

dlink-admin/src/main/java/com/dlink/interceptor/TenantInterceptor.java

@@ -19,7 +19,7 @@
 
 package com.dlink.interceptor;
 
-import com.dlink.context.RequestContext;
+import com.dlink.context.TenantContextHolder;
 
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
@@ -41,7 +41,7 @@ public class TenantInterceptor implements HandlerInterceptor {
                              Object handler) throws Exception {
         String tenantId = request.getHeader("tenantId");
         if (!StringUtils.isNullOrEmpty(tenantId)) {
-            RequestContext.set(Integer.valueOf(tenantId));
+            TenantContextHolder.set(Integer.valueOf(tenantId));
         }
         return HandlerInterceptor.super.preHandle(request, response, handler);
     }

dlink-admin/src/main/java/com/dlink/mapper/JobHistoryMapper.java

@@ -24,6 +24,8 @@ import com.dlink.model.JobHistory;
 
 import org.apache.ibatis.annotations.Mapper;
 
+import com.baomidou.mybatisplus.annotation.InterceptorIgnore;
+
 /**
  * JobHistoryMapper
  *
@@ -33,6 +35,9 @@ import org.apache.ibatis.annotations.Mapper;
 @Mapper
 public interface JobHistoryMapper extends SuperMapper<JobHistory> {
 
+    @InterceptorIgnore(tenantLine = "true")
+    JobHistory getByIdWithoutTenant(Integer id);
+
     int insert(JobHistory jobHistory);
 
 }

dlink-admin/src/main/java/com/dlink/mapper/JobInstanceMapper.java

@@ -27,6 +27,8 @@ import org.apache.ibatis.annotations.Mapper;
 
 import java.util.List;
 
+import com.baomidou.mybatisplus.annotation.InterceptorIgnore;
+
 /**
  * JobInstanceMapper
  *
@@ -36,6 +38,9 @@ import java.util.List;
 @Mapper
 public interface JobInstanceMapper extends SuperMapper<JobInstance> {
 
+    @InterceptorIgnore(tenantLine = "true")
+    JobInstance getByIdWithoutTenant(Integer id);
+
     List<JobInstanceCount> countStatus();
 
     List<JobInstanceCount> countHistoryStatus();

dlink-admin/src/main/java/com/dlink/mapper/TaskMapper.java

@@ -27,6 +27,8 @@ import org.apache.ibatis.annotations.Param;
 
 import java.util.List;
 
+import com.baomidou.mybatisplus.annotation.InterceptorIgnore;
+
 /**
  * 作业 Mapper 接口
  *
@@ -38,9 +40,11 @@ public interface TaskMapper extends SuperMapper<Task> {
 
     Integer queryAllSizeByName(String name);
 
-    List<Task> queryOnLineTaskByDoneStatus(@Param("parentIds") List<Integer> parentIds
-            , @Param("stepIds") List<Integer> stepIds, @Param("includeNull") boolean includeNull
-            , @Param("jobStatuses") List<String> jobStatuses);
+    List<Task> queryOnLineTaskByDoneStatus(@Param("parentIds") List<Integer> parentIds,
+                                           @Param("stepIds") List<Integer> stepIds,
+                                           @Param("includeNull") boolean includeNull,
+                                           @Param("jobStatuses") List<String> jobStatuses);
 
+    @InterceptorIgnore(tenantLine = "true")
     Task getTaskByNameAndTenantId(@Param("name") String name, @Param("tenantId") Integer tenantId);
 }

dlink-admin/src/main/java/com/dlink/service/JobHistoryService.java

@@ -30,9 +30,12 @@ import com.dlink.model.JobHistory;
  **/
 public interface JobHistoryService extends ISuperService<JobHistory> {
 
+    JobHistory getByIdWithoutTenant(Integer id);
+
     JobHistory getJobHistory(Integer id);
 
     JobHistory getJobHistoryInfo(JobHistory jobHistory);
 
     JobHistory refreshJobHistory(Integer id, String jobManagerHost, String jobId, boolean needSave);
+
 }

dlink-admin/src/main/java/com/dlink/service/JobInstanceService.java

@@ -38,6 +38,8 @@ import com.fasterxml.jackson.databind.JsonNode;
  */
 public interface JobInstanceService extends ISuperService<JobInstance> {
 
+    JobInstance getByIdWithoutTenant(Integer id);
+
     JobInstanceStatus getStatusCount(boolean isHistory);
 
     List<JobInstance> listJobInstanceActive();

dlink-admin/src/main/java/com/dlink/service/impl/JobHistoryServiceImpl.java

@@ -43,6 +43,11 @@ import com.fasterxml.jackson.databind.JsonNode;
 @Service
 public class JobHistoryServiceImpl extends SuperServiceImpl<JobHistoryMapper, JobHistory> implements JobHistoryService {
 
+    @Override
+    public JobHistory getByIdWithoutTenant(Integer id) {
+        return baseMapper.getByIdWithoutTenant(id);
+    }
+
     @Override
     public JobHistory getJobHistory(Integer id) {
         return getJobHistoryInfo(getById(id));
@@ -111,11 +116,10 @@ public class JobHistoryServiceImpl extends SuperServiceImpl<JobHistoryMapper, Jo
             jobHistory.setCheckpointsConfigJson(JSONUtil.toJsonString(checkPointsConfig));
             jobHistory.setConfigJson(JSONUtil.toJsonString(jobsConfig));
             if (needSave) {
-                if (Asserts.isNotNull(getById(id))) {
                 updateById(jobHistory);
-                } else {
-                    save(jobHistory);
-                }
+                /*
+                 * if (Asserts.isNotNull(getById(id))) { updateById(jobHistory); } else { save(jobHistory); }
+                 */
             }
         } catch (Exception e) {
             e.printStackTrace();

dlink-admin/src/main/java/com/dlink/service/impl/JobInstanceServiceImpl.java

@@ -59,7 +59,9 @@ import com.fasterxml.jackson.databind.ObjectMapper;
  * @since 2022/2/2 13:52
  */
 @Service
-public class JobInstanceServiceImpl extends SuperServiceImpl<JobInstanceMapper, JobInstance> implements JobInstanceService {
+public class JobInstanceServiceImpl extends SuperServiceImpl<JobInstanceMapper, JobInstance>
+        implements
+            JobInstanceService {
 
     @Autowired
     private HistoryService historyService;
@@ -70,6 +72,11 @@ public class JobInstanceServiceImpl extends SuperServiceImpl<JobInstanceMapper,
     @Autowired
     private JobHistoryService jobHistoryService;
 
+    @Override
+    public JobInstance getByIdWithoutTenant(Integer id) {
+        return baseMapper.getByIdWithoutTenant(id);
+    }
+
     @Override
     public JobInstanceStatus getStatusCount(boolean isHistory) {
         List<JobInstanceCount> jobInstanceCounts = null;
@@ -153,7 +160,8 @@ public class JobInstanceServiceImpl extends SuperServiceImpl<JobInstanceMapper,
             history.setConfig(JSONUtil.parseObject(history.getConfigJson()));
             jobInfoDetail.setHistory(history);
             if (Asserts.isNotNull(history.getClusterConfigurationId())) {
-                jobInfoDetail.setClusterConfiguration(clusterConfigurationService.getClusterConfigById(history.getClusterConfigurationId()));
+                jobInfoDetail.setClusterConfiguration(
+                        clusterConfigurationService.getClusterConfigById(history.getClusterConfigurationId()));
             }
             return jobInfoDetail;
         }
@@ -174,7 +182,8 @@ public class JobInstanceServiceImpl extends SuperServiceImpl<JobInstanceMapper,
         history.setConfig(JSONUtil.parseObject(history.getConfigJson()));
         jobInfoDetail.setHistory(history);
         if (Asserts.isNotNull(history) && Asserts.isNotNull(history.getClusterConfigurationId())) {
-            jobInfoDetail.setClusterConfiguration(clusterConfigurationService.getClusterConfigById(history.getClusterConfigurationId()));
+            jobInfoDetail.setClusterConfiguration(
+                    clusterConfigurationService.getClusterConfigById(history.getClusterConfigurationId()));
         }
         if (pool.exist(key)) {
             pool.refresh(jobInfoDetail);
@@ -219,7 +228,8 @@ public class JobInstanceServiceImpl extends SuperServiceImpl<JobInstanceMapper,
                 list.get(i).setDuration(pool.get(list.get(i).getId().toString()).getInstance().getDuration());
             }
         }
-        return ProTableResult.<JobInstance>builder().success(true).data(list).total(page.getTotal()).current(current).pageSize(pageSize).build();
+        return ProTableResult.<JobInstance>builder().success(true).data(list).total(page.getTotal()).current(current)
+                .pageSize(pageSize).build();
     }
 
 }

dlink-admin/src/main/java/com/dlink/service/impl/StudioServiceImpl.java

@@ -177,7 +177,7 @@ public class StudioServiceImpl implements StudioService {
         JobConfig config = studioExecuteDTO.getJobConfig();
         buildSession(config);
         // To initialize java udf, but it only support local mode.
-        UDFPath udfPath = udfService.initUDF(studioExecuteDTO.getStatement(), config.getGatewayConfig() == null ? null : config.getGatewayConfig().getType());
+        UDFPath udfPath = udfService.initUDF(studioExecuteDTO.getStatement(), GatewayType.get(config.getType()));
         config.setJarFiles(udfPath.getJarPaths());
         config.setPyFiles(udfPath.getPyPaths());
         JobManager jobManager = JobManager.build(config);

dlink-admin/src/main/java/com/dlink/service/impl/TaskServiceImpl.java

@@ -31,6 +31,7 @@ import com.dlink.common.result.Result;
 import com.dlink.config.Dialect;
 import com.dlink.constant.FlinkRestResultConstant;
 import com.dlink.constant.NetConstant;
+import com.dlink.context.TenantContextHolder;
 import com.dlink.daemon.task.DaemonFactory;
 import com.dlink.daemon.task.DaemonTaskConfig;
 import com.dlink.db.service.impl.SuperServiceImpl;
@@ -209,7 +210,7 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
                     task.getDatabaseId(), null));
         }
         JobConfig config = buildJobConfig(task);
-        UDFPath udfPath = udfService.initUDF(task.getStatement(), config.getGatewayConfig().getType());
+        UDFPath udfPath = udfService.initUDF(task.getStatement(), GatewayType.get(config.getType()));
         config.setJarFiles(udfPath.getJarPaths());
         config.setPyFiles(udfPath.getPyPaths());
         JobManager jobManager = JobManager.build(config);
@@ -371,7 +372,8 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
                 Map<String, String> config = task.getConfig().get(0);
                 UDFTemplate template = udfTemplateService.getById(config.get("templateId"));
                 if (template != null) {
-                    String code = UDFUtil.templateParse(task.getDialect(), template.getTemplateCode(), config.get("className"));
+                    String code = UDFUtil.templateParse(task.getDialect(), template.getTemplateCode(),
+                            config.get("className"));
                     task.setStatement(code);
                 }
             }
@@ -441,6 +443,8 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
         Task defaultFlinkSQLEnvTask = getTaskByNameAndTenantId(name, tenantId);
         if (null == defaultFlinkSQLEnvTask) {
             defaultFlinkSQLEnvTask = new Task();
+        } else {
+            return defaultFlinkSQLEnvTask;
         }
         // defaultFlinkSQLEnvTask.setId(1);
         defaultFlinkSQLEnvTask.setName(name);
@@ -502,7 +506,8 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
     @Override
     public Task getUDFByClassName(String className) {
         Task task = getOne(
-            new QueryWrapper<Task>().in("dialect", Dialect.JAVA, Dialect.SCALA, Dialect.PYTHON).eq("enabled", 1).eq("save_point_path", className));
+                new QueryWrapper<Task>().in("dialect", Dialect.JAVA, Dialect.SCALA, Dialect.PYTHON).eq("enabled", 1)
+                        .eq("save_point_path", className));
         Assert.check(task);
         task.setStatement(statementService.getById(task.getId()).getStatement());
         return task;
@@ -511,7 +516,8 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
     @Override
     public List<Task> getAllUDF() {
         List<Task> tasks =
-            list(new QueryWrapper<Task>().in("dialect", Dialect.JAVA, Dialect.SCALA, Dialect.PYTHON).eq("enabled", 1).isNotNull("save_point_path"));
+                list(new QueryWrapper<Task>().in("dialect", Dialect.JAVA, Dialect.SCALA, Dialect.PYTHON)
+                        .eq("enabled", 1).isNotNull("save_point_path"));
         return tasks.stream().peek(task -> {
             Assert.check(task);
             task.setStatement(statementService.getById(task.getId()).getStatement());
@@ -858,8 +864,9 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
             jobInfoDetail = pool.get(key);
         } else {
             jobInfoDetail = new JobInfoDetail(id);
-            JobInstance jobInstance = jobInstanceService.getById(id);
+            JobInstance jobInstance = jobInstanceService.getByIdWithoutTenant(id);
             Asserts.checkNull(jobInstance, "该任务实例不存在");
+            TenantContextHolder.set(jobInstance.getTenantId());
             jobInfoDetail.setInstance(jobInstance);
             Cluster cluster = clusterService.getById(jobInstance.getClusterId());
             jobInfoDetail.setCluster(cluster);

dlink-admin/src/main/java/com/dlink/service/impl/TenantServiceImpl.java

@@ -21,7 +21,7 @@ package com.dlink.service.impl;
 
 import com.dlink.assertion.Asserts;
 import com.dlink.common.result.Result;
-import com.dlink.context.RequestContext;
+import com.dlink.context.TenantContextHolder;
 import com.dlink.db.service.impl.SuperServiceImpl;
 import com.dlink.mapper.TenantMapper;
 import com.dlink.model.Namespace;
@@ -66,7 +66,7 @@ public class TenantServiceImpl extends SuperServiceImpl<TenantMapper, Tenant> im
             }
             tenant.setIsDelete(false);
             if (save(tenant)) {
-                RequestContext.set(tenant.getId());
+                TenantContextHolder.set(tenant.getId());
                 return Result.succeed("新增成功");
             }
             return Result.failed("新增失败");
@@ -101,12 +101,14 @@ public class TenantServiceImpl extends SuperServiceImpl<TenantMapper, Tenant> im
                 return Result.failed("租户不存在");
             }
 
-            Long tenantRoleCount = roleService.getBaseMapper().selectCount(new QueryWrapper<Role>().eq("tenant_id", id));
+            Long tenantRoleCount =
+                    roleService.getBaseMapper().selectCount(new QueryWrapper<Role>().eq("tenant_id", id));
             if (tenantRoleCount > 0) {
                 return Result.failed("删除租户失败，该租户已绑定角色");
             }
 
-            Long tenantNamespaceCount = namespaceService.getBaseMapper().selectCount(new QueryWrapper<Namespace>().eq("tenant_id", id));
+            Long tenantNamespaceCount =
+                    namespaceService.getBaseMapper().selectCount(new QueryWrapper<Namespace>().eq("tenant_id", id));
             if (tenantNamespaceCount > 0) {
                 return Result.failed("删除租户失败，该租户已绑定名称空间");
             }
@@ -162,8 +164,8 @@ public class TenantServiceImpl extends SuperServiceImpl<TenantMapper, Tenant> im
     public Result switchTenant(JsonNode para) {
         if (para.size() > 0) {
             Integer tenantId = para.get("tenantId").asInt();
-            RequestContext.remove();
-            RequestContext.set(tenantId);
+            TenantContextHolder.clear();
+            TenantContextHolder.set(tenantId);
             return Result.succeed("切换租户成功");
         } else {
             return Result.failed("无法切换租户,获取不到租户信息");

dlink-admin/src/main/java/com/dlink/service/impl/UserServiceImpl.java

@@ -21,7 +21,7 @@ package com.dlink.service.impl;
 
 import com.dlink.assertion.Asserts;
 import com.dlink.common.result.Result;
-import com.dlink.context.RequestContext;
+import com.dlink.context.TenantContextHolder;
 import com.dlink.db.service.impl.SuperServiceImpl;
 import com.dlink.dto.LoginUTO;
 import com.dlink.dto.UserDTO;
@@ -146,7 +146,7 @@ public class UserServiceImpl extends SuperServiceImpl<UserMapper, User> implemen
             }
 
             // 将前端入参 租户id 放入上下文
-            RequestContext.set(loginUTO.getTenantId());
+            TenantContextHolder.set(loginUTO.getTenantId());
 
             // get user tenants and roles
             UserDTO userDTO = getUserALLBaseInfo(loginUTO, user);
@@ -177,7 +177,8 @@ public class UserServiceImpl extends SuperServiceImpl<UserMapper, User> implemen
         });
 
         userTenants.forEach(userTenant -> {
-            Tenant tenant = tenantService.getBaseMapper().selectOne(new QueryWrapper<Tenant>().eq("id", userTenant.getTenantId()));
+            Tenant tenant = tenantService.getBaseMapper()
+                    .selectOne(new QueryWrapper<Tenant>().eq("id", userTenant.getTenantId()));
             if (Asserts.isNotNull(tenant)) {
                 tenantList.add(tenant);
             }

dlink-admin/src/main/resources/mapper/JobHistoryMapper.xml

@@ -2,6 +2,12 @@
 <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
 <mapper namespace="com.dlink.mapper.JobHistoryMapper">
 
+    <select id="getByIdWithoutTenant" resultType="com.dlink.model.JobHistory">
+        select *
+        from dlink_job_history
+        where id = #{id}
+        limit 1
+    </select>
 
     <insert id="insert">
         insert into dlink_job_history (id,job_json,exceptions_json,checkpoints_json,checkpoints_config_json,config_json,

dlink-admin/src/main/resources/mapper/JobInstanceMapper.xml

@@ -2,6 +2,13 @@
 <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
 <mapper namespace="com.dlink.mapper.JobInstanceMapper">
 
+    <select id="getByIdWithoutTenant" resultType="com.dlink.model.JobInstance">
+        select *
+        from dlink_job_instance
+        where id = #{id}
+        limit 1
+    </select>
+
     <select id="selectForProTable" resultType="com.dlink.model.JobInstance">
         select
         a.*,

dlink-admin/src/main/resources/mapper/TaskMapper.xml

@@ -59,8 +59,8 @@
         select *
         from dlink_task
         where 1 = 1
-          and name = "${name}"
-          and tenant_id = ${tenantId}
+          and name = #{name}
+          and tenant_id = #{tenantId}
     </select>
 
     <select id="queryOnLineTaskByDoneStatus" resultType="com.dlink.model.Task">

dlink-client/dlink-client-1.14/pom.xml

@@ -15,16 +15,14 @@
   ~ See the License for the specific language governing permissions and
   ~ limitations under the License.
   -->
-
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
     <parent>
-        <artifactId>dlink-client</artifactId>
         <groupId>com.dlink</groupId>
+        <artifactId>dlink-client</artifactId>
         <version>0.6.8-SNAPSHOT</version>
     </parent>
-    <modelVersion>4.0.0</modelVersion>
 
     <artifactId>dlink-client-1.14</artifactId>
     <properties>
@@ -54,5 +52,11 @@
             <artifactId>jackson-datatype-jsr310</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>com.starrocks</groupId>
+            <artifactId>flink-connector-starrocks</artifactId>
+            <version>1.2.3_flink-1.14_2.12</version>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 </project>

dlink-flink/dlink-flink-1.14/pom.xml

@@ -15,16 +15,14 @@
   ~ See the License for the specific language governing permissions and
   ~ limitations under the License.
   -->
-
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
     <parent>
-        <artifactId>dlink-flink</artifactId>
         <groupId>com.dlink</groupId>
+        <artifactId>dlink-flink</artifactId>
         <version>0.6.8-SNAPSHOT</version>
     </parent>
-    <modelVersion>4.0.0</modelVersion>
 
     <artifactId>dlink-flink-1.14</artifactId>
 
@@ -45,17 +43,19 @@
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
             <exclusions>
                 <exclusion>
                     <groupId>org.slf4j</groupId>
                     <artifactId>slf4j-api</artifactId>
                 </exclusion>
             </exclusions>
-            <version>${flink.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-runtime_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
             <exclusions>
                 <exclusion>
                     <groupId>org.slf4j</groupId>
@@ -66,8 +66,6 @@
                     <artifactId>flink-shaded-guava</artifactId>
                 </exclusion>
             </exclusions>
-            <version>${flink.version}</version>
-            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
@@ -77,13 +75,13 @@
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-clients_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
             <exclusions>
                 <exclusion>
                     <groupId>org.slf4j</groupId>
                     <artifactId>slf4j-api</artifactId>
                 </exclusion>
             </exclusions>
-            <version>${flink.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
@@ -143,8 +141,8 @@
             <version>${flinkcdc.version}</version>
             <exclusions>
                 <exclusion>
-                    <artifactId>flink-shaded-guava</artifactId>
                     <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-shaded-guava</artifactId>
                 </exclusion>
             </exclusions>
         </dependency>
@@ -183,10 +181,5 @@
             <artifactId>flink-doris-connector-1.14_${scala.binary.version}</artifactId>
             <version>1.1.0</version>
         </dependency>
-        <dependency>
-            <groupId>com.starrocks</groupId>
-            <artifactId>flink-connector-starrocks</artifactId>
-            <version>1.2.3_flink-1.14_${scala.binary.version}</version>
-        </dependency>
     </dependencies>
 </project>

dlink-process/src/main/java/com/dlink/process/pool/ConsolePool.java

@@ -55,7 +55,8 @@ public class ConsolePool extends AbstractPool<StringBuilder> {
         if (consoleEntityMap.containsKey(user)) {
             consoleEntityMap.get(user).append(str);
         } else {
-            consoleEntityMap.put(user, new StringBuilder(str));
+            StringBuilder sb = new StringBuilder("Dinky User Console:");
+            consoleEntityMap.put(user, sb.append(str));
         }
     }