单元测试：ClusterHostTaskServiceTest，ClusterRoleTaskServiceTest,ClusterTaskServiceTest,KafkaFileServiceTest,TopicCommandsTest,TopicReassignUtilsTest

kafka-manager-core/src/test/java/com/xiaojukeji/kafka/manager/service/service/AdminServiceTest.java

@@ -0,0 +1,375 @@
+package com.xiaojukeji.kafka.manager.service.service;
+
+import com.xiaojukeji.kafka.manager.common.bizenum.TaskStatusEnum;
+import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
+import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterDO;
+import com.xiaojukeji.kafka.manager.common.entity.pojo.TopicDO;
+import com.xiaojukeji.kafka.manager.service.config.BaseTest;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.Arrays;
+import java.util.Date;
+import java.util.Properties;
+
+/**
+ * @author xuguang
+ * @Date 2021/12/24
+ */
+public class AdminServiceTest extends BaseTest {
+
+    /**
+     * 集群共包括三个broker:1,2,3, 该topic 1分区 1副本因子，在broker1上
+     */
+    private final static String REAL_TOPIC1_IN_ZK = "moduleTest";
+
+    /**
+     * 集群共包括三个broker:1,2,3, 该topic 2分区 3副本因子，在broker1,2,3上
+     */
+    private final static String REAL_TOPIC2_IN_ZK = "xgTest";
+
+    private final static String INVALID_TOPIC = "xxxxx";
+
+    private final static String ZK_DEFAULT_TOPIC = "_consumer_offsets";
+
+    private final static String CREATE_TOPIC_TEST = "createTopicTest";
+
+    private final static Long REAL_CLUSTER_ID_IN_MYSQL = 1L;
+
+    private final static Integer REAL_BROKER_ID_IN_ZK = 1;
+
+    private final static Long INVALID_CLUSTER_ID = -1L;
+
+    private final static Integer INVALID_PARTITION_ID = -1;
+
+    private final static Integer REAL_PARTITION_ID = 0;
+
+    private final static Integer INVALID_BROKER_ID = -1;
+
+    private final static String APP_ID = "dkm_admin";
+
+    private final static Long INVALID_REGION_ID = -1L;
+
+    private final static Long REAL_REGION_ID_IN_MYSQL = 1L;
+
+    private final static String ADMIN = "admin";
+
+
+    @Autowired
+    private AdminService adminService;
+
+    @Autowired
+    private TopicManagerService topicManagerService;
+
+    private TopicDO getTopicDO() {
+        TopicDO topicDO = new TopicDO();
+        topicDO.setClusterId(REAL_CLUSTER_ID_IN_MYSQL);
+        topicDO.setTopicName(CREATE_TOPIC_TEST);
+        topicDO.setAppId(APP_ID);
+        topicDO.setDescription(CREATE_TOPIC_TEST);
+        topicDO.setPeakBytesIn(100000L);
+        return topicDO;
+    }
+
+    public ClusterDO getClusterDO() {
+        ClusterDO clusterDO = new ClusterDO();
+        clusterDO.setId(REAL_CLUSTER_ID_IN_MYSQL);
+        clusterDO.setClusterName("LogiKM_moduleTest");
+        clusterDO.setZookeeper("10.190.46.198:2181,10.190.14.237:2181,10.190.50.65:2181/xg");
+        clusterDO.setBootstrapServers("10.190.46.198:9093,10.190.14.237:9093,10.190.50.65:9093");
+        clusterDO.setSecurityProperties("{ \t\"security.protocol\": \"SASL_PLAINTEXT\", \t\"sasl.mechanism\": \"PLAIN\", \t\"sasl.jaas.config\": \"org.apache.kafka.common.security.plain.PlainLoginModule required username=\\\"dkm_admin\\\" password=\\\"km_kMl4N8as1Kp0CCY\\\";\" }");
+        clusterDO.setStatus(1);
+        clusterDO.setGmtCreate(new Date());
+        clusterDO.setGmtModify(new Date());
+        return clusterDO;
+    }
+
+    @Test(description = "测试创建topic")
+    public void createTopicTest() {
+        // broker not exist
+        createTopic2BrokerNotExistTest();
+        // success to create topic
+        createTopic2SuccessTest();
+        // failure to create topic, topic already exists
+        createTopic2FailureTest();
+    }
+
+    private void createTopic2BrokerNotExistTest() {
+        TopicDO topicDO = getTopicDO();
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus result = adminService.createTopic(
+                clusterDO,
+                topicDO,
+                1,
+                1,
+                1L,
+                Arrays.asList(INVALID_BROKER_ID),
+                new Properties(),
+                ADMIN,
+                ADMIN);
+        Assert.assertEquals(result.getCode(), ResultStatus.BROKER_NOT_EXIST.getCode());
+    }
+
+    private void createTopic2FailureTest() {
+        TopicDO topicDO = getTopicDO();
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus result = adminService.createTopic(
+                clusterDO,
+                topicDO,
+                1,
+                1,
+                INVALID_REGION_ID,
+                Arrays.asList(REAL_BROKER_ID_IN_ZK),
+                new Properties(),
+                ADMIN,
+                ADMIN);
+        Assert.assertNotEquals(result.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    private void createTopic2SuccessTest() {
+        TopicDO topicDO = getTopicDO();
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus result = adminService.createTopic(
+                clusterDO,
+                topicDO,
+                1,
+                1,
+                INVALID_REGION_ID,
+                Arrays.asList(REAL_BROKER_ID_IN_ZK),
+                new Properties(),
+                ADMIN,
+                ADMIN);
+        Assert.assertEquals(result.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    @Test(description = "测试删除topic")
+    public void deleteTopicTest() {
+        // topic does not exist
+        deleteTopic2FailureTest();
+        // success to delete
+        deleteTopic2SuccessTest();
+    }
+
+    private void deleteTopic2FailureTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus resultStatus = adminService.deleteTopic(
+                clusterDO,
+                INVALID_TOPIC,
+                ADMIN
+        );
+        Assert.assertNotEquals(resultStatus.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    private void deleteTopic2SuccessTest() {
+        TopicDO topicDO = getTopicDO();
+        topicManagerService.addTopic(topicDO);
+
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus resultStatus = adminService.deleteTopic(
+                clusterDO,
+                CREATE_TOPIC_TEST,
+                ADMIN
+        );
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    @Test(description = "测试优先副本选举状态")
+    public void preferredReplicaElectionStatusTest() {
+        // running
+        preferredReplicaElectionStatus2RunningTest();
+        // not running
+        preferredReplicaElectionStatus2NotRunningTest();
+    }
+
+    private void preferredReplicaElectionStatus2RunningTest() {
+        // zk上需要创建/admin/preferred_replica_election节点
+        ClusterDO clusterDO = getClusterDO();
+        TaskStatusEnum taskStatusEnum = adminService.preferredReplicaElectionStatus(clusterDO);
+        Assert.assertEquals(taskStatusEnum.getCode(), TaskStatusEnum.RUNNING.getCode());
+    }
+
+    private void preferredReplicaElectionStatus2NotRunningTest() {
+        ClusterDO clusterDO = getClusterDO();
+        // zk上无/admin/preferred_replica_election节点
+        TaskStatusEnum taskStatusEnum = adminService.preferredReplicaElectionStatus(clusterDO);
+        Assert.assertEquals(taskStatusEnum.getCode(), TaskStatusEnum.SUCCEED.getCode());
+    }
+
+    @Test(description = "测试集群纬度优先副本选举")
+    public void preferredReplicaElectionOfCluster2Test() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus resultStatus = adminService.preferredReplicaElection(clusterDO, ADMIN);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    @Test(description = "Broker纬度优先副本选举")
+    public void preferredReplicaElectionOfBrokerTest() {
+        // 参数异常
+        preferredReplicaElectionOfBroker2ParamIllegalTest();
+        // success
+        preferredReplicaElectionOfBroker2SuccessTest();
+    }
+
+    private void preferredReplicaElectionOfBroker2ParamIllegalTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus resultStatus = adminService.preferredReplicaElection(
+                clusterDO,
+                INVALID_BROKER_ID,
+                ADMIN
+        );
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.PARAM_ILLEGAL.getCode());
+    }
+
+    private void preferredReplicaElectionOfBroker2SuccessTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus resultStatus = adminService.preferredReplicaElection(
+                clusterDO,
+                REAL_BROKER_ID_IN_ZK,
+                ADMIN
+        );
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    @Test(description = "Topic纬度优先副本选举")
+    public void preferredReplicaElectionOfTopicTest() {
+        // topic not exist
+        preferredReplicaElectionOfTopic2TopicNotExistTest();
+        // success
+        preferredReplicaElectionOfTopic2SuccessTest();
+    }
+
+    private void preferredReplicaElectionOfTopic2TopicNotExistTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus resultStatus = adminService.preferredReplicaElection(
+                clusterDO,
+                INVALID_TOPIC,
+                ADMIN
+        );
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.TOPIC_NOT_EXIST.getCode());
+    }
+
+    private void preferredReplicaElectionOfTopic2SuccessTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus resultStatus = adminService.preferredReplicaElection(
+                clusterDO,
+                REAL_TOPIC1_IN_ZK,
+                ADMIN
+        );
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    @Test(description = "Topic纬度优先副本选举")
+    public void preferredReplicaElectionOfPartitionTest() {
+        // topic not exist
+        preferredReplicaElectionOfPartition2TopicNotExistTest();
+        // partition Not Exist
+        preferredReplicaElectionOfPartition2PartitionNotExistTest();
+        // success
+        preferredReplicaElectionOfPartition2SuccessTest();
+    }
+
+    private void preferredReplicaElectionOfPartition2TopicNotExistTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus resultStatus = adminService.preferredReplicaElection(
+                clusterDO,
+                INVALID_TOPIC,
+                INVALID_PARTITION_ID,
+                ADMIN
+        );
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.TOPIC_NOT_EXIST.getCode());
+    }
+
+    private void preferredReplicaElectionOfPartition2PartitionNotExistTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus resultStatus = adminService.preferredReplicaElection(
+                clusterDO,
+                REAL_TOPIC2_IN_ZK,
+                INVALID_PARTITION_ID,
+                ADMIN
+        );
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.PARTITION_NOT_EXIST.getCode());
+    }
+
+    private void preferredReplicaElectionOfPartition2SuccessTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus resultStatus = adminService.preferredReplicaElection(
+                clusterDO,
+                REAL_TOPIC2_IN_ZK,
+                REAL_PARTITION_ID,
+                ADMIN
+        );
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    @Test(description = "测试获取Topic配置")
+    public void getTopicConfigTest() {
+        // result is null
+        getTopicConfig2NullTest();
+        // result not null
+        getTopicConfig2NotNullTest();
+    }
+
+    private void getTopicConfig2NullTest() {
+        ClusterDO clusterDO = getClusterDO();
+        clusterDO.setId(INVALID_CLUSTER_ID);
+        Properties topicConfig = adminService.getTopicConfig(clusterDO, REAL_TOPIC1_IN_ZK);
+        Assert.assertNull(topicConfig);
+    }
+
+    private void getTopicConfig2NotNullTest() {
+        ClusterDO clusterDO = getClusterDO();
+        Properties topicConfig = adminService.getTopicConfig(clusterDO, REAL_TOPIC1_IN_ZK);
+        Assert.assertNotNull(topicConfig);
+    }
+
+    @Test(description = "测试修改Topic配置")
+    public void modifyTopicConfigTest() {
+        ClusterDO clusterDO = getClusterDO();
+        Properties properties = new Properties();
+        properties.put("retention.ms", "21600000");
+        ResultStatus resultStatus = adminService.modifyTopicConfig(
+                clusterDO,
+                REAL_TOPIC1_IN_ZK,
+                properties,
+                ADMIN
+        );
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    @Test(description = "测试扩分区")
+    public void expandPartitionsTest() {
+        // broker not exist
+        expandPartitions2BrokerNotExistTest();
+        // success
+        expandPartitions2SuccessTest();
+    }
+
+    private void expandPartitions2BrokerNotExistTest() {
+        // 存在两个下线broker, region中包含一个
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus resultStatus = adminService.expandPartitions(
+                clusterDO,
+                REAL_TOPIC1_IN_ZK,
+                2,
+                REAL_REGION_ID_IN_MYSQL,
+                Arrays.asList(INVALID_BROKER_ID),
+                ADMIN
+        );
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.BROKER_NOT_EXIST.getCode());
+    }
+
+    private void expandPartitions2SuccessTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus resultStatus = adminService.expandPartitions(
+                clusterDO,
+                REAL_TOPIC1_IN_ZK,
+                2,
+                INVALID_REGION_ID,
+                Arrays.asList(REAL_BROKER_ID_IN_ZK),
+                ADMIN
+        );
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.BROKER_NOT_EXIST.getCode());
+    }
+
+}

kafka-manager-core/src/test/java/com/xiaojukeji/kafka/manager/service/service/ExpertServiceTest.java

@@ -0,0 +1,80 @@
+package com.xiaojukeji.kafka.manager.service.service;
+
+import com.xiaojukeji.kafka.manager.common.entity.ao.config.expert.RegionTopicHotConfig;
+import com.xiaojukeji.kafka.manager.common.entity.ao.expert.TopicRegionHot;
+import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterDO;
+import com.xiaojukeji.kafka.manager.service.config.BaseTest;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+
+/**
+ * @author xuguang
+ * Collections.EmptyList()
+ * @Date 2021/12/24
+ */
+public class ExpertServiceTest extends BaseTest {
+
+    private final static Long REAL_CLUSTER_ID_IN_MYSQL = 1L;
+
+    private final static Long INVALID_CLUSTER_ID = -1L;
+
+    @Autowired
+    @InjectMocks
+    private ExpertService expertService;
+
+    @Mock
+    private ConfigService configService;
+
+    @Mock
+    private ClusterService clusterService;
+
+    @BeforeMethod
+    public void setup() {
+        MockitoAnnotations.initMocks(this);
+    }
+
+    private RegionTopicHotConfig getRegionTopicHotConfig() {
+        RegionTopicHotConfig regionTopicHotConfig = new RegionTopicHotConfig();
+        regionTopicHotConfig.setMaxDisPartitionNum(5);
+        regionTopicHotConfig.setIgnoreClusterIdList(Arrays.asList(INVALID_CLUSTER_ID));
+        return regionTopicHotConfig;
+    }
+
+    public ClusterDO getClusterDO() {
+        ClusterDO clusterDO = new ClusterDO();
+        clusterDO.setId(REAL_CLUSTER_ID_IN_MYSQL);
+        clusterDO.setClusterName("LogiKM_moduleTest");
+        clusterDO.setZookeeper("10.190.46.198:2181,10.190.14.237:2181,10.190.50.65:2181/xg");
+        clusterDO.setBootstrapServers("10.190.46.198:9093,10.190.14.237:9093,10.190.50.65:9093");
+        clusterDO.setSecurityProperties("{ \t\"security.protocol\": \"SASL_PLAINTEXT\", \t\"sasl.mechanism\": \"PLAIN\", \t\"sasl.jaas.config\": \"org.apache.kafka.common.security.plain.PlainLoginModule required username=\\\"dkm_admin\\\" password=\\\"km_kMl4N8as1Kp0CCY\\\";\" }");
+        clusterDO.setStatus(1);
+        clusterDO.setGmtCreate(new Date());
+        clusterDO.setGmtModify(new Date());
+        return clusterDO;
+    }
+
+    @Test(description = "测试Region内热点Topic")
+    public void getRegionHotTopics() {
+        RegionTopicHotConfig regionTopicHotConfig = getRegionTopicHotConfig();
+        Mockito.when(configService.getByKey(Mockito.anyString(), Mockito.any())).thenReturn(regionTopicHotConfig);
+        ClusterDO clusterDO1 = getClusterDO();
+        ClusterDO clusterDO2 = getClusterDO();
+        clusterDO2.setId(INVALID_CLUSTER_ID);
+        Mockito.when(clusterService.list()).thenReturn(Arrays.asList(clusterDO1, clusterDO2));
+
+        List<TopicRegionHot> regionHotTopics = expertService.getRegionHotTopics();
+        Assert.assertFalse(regionHotTopics.isEmpty());
+    }
+
+}

kafka-manager-core/src/test/java/com/xiaojukeji/kafka/manager/service/service/gateway/AppServiceTest.java

@@ -27,13 +27,13 @@ public class AppServiceTest extends BaseTest {
     public static Object[][] provideAppDO() {
         AppDO appDO = new AppDO();
         appDO.setId(4L);
-        appDO.setAppId("testAppId");
+        appDO.setAppId("moduleTestAppId");
-        appDO.setName("testApp");
+        appDO.setName("moduleTestApp");
-        appDO.setPassword("testApp");
+        appDO.setPassword("moduleTestApp");
         appDO.setType(1);
         appDO.setApplicant("admin");
         appDO.setPrincipals("admin");
-        appDO.setDescription("testApp");
+        appDO.setDescription("moduleTestApp");
         appDO.setCreateTime(new Date(1638786493173L));
         appDO.setModifyTime(new Date(1638786493173L));
         return new Object[][] {{appDO}};
@@ -59,7 +59,6 @@ public class AppServiceTest extends BaseTest {
         addApp2MysqlErrorTest();
     }
 
-    @Rollback(false)
     private void addApp2SuccessTest(AppDO appDO) {
         ResultStatus addAppResult = appService.addApp(appDO, "admin");
         Assert.assertEquals(addAppResult.getCode(), ResultStatus.SUCCESS.getCode());
@@ -70,7 +69,6 @@ public class AppServiceTest extends BaseTest {
         Assert.assertEquals(addAppResult.getCode(), ResultStatus.RESOURCE_ALREADY_EXISTED.getCode());
     }
 
-    @Rollback()
     private void addApp2MysqlErrorTest() {
         ResultStatus addAppResult = appService.addApp(new AppDO(), "admin");
         Assert.assertEquals(addAppResult.getCode(), ResultStatus.MYSQL_ERROR.getCode());
@@ -78,20 +76,24 @@ public class AppServiceTest extends BaseTest {
 
     @Test(dataProvider = "provideAppDO")
     public void deleteAppTest(AppDO appDO) {
+        appService.addApp(appDO, "admin");
+
         // 测试删除app成功
         deleteApp2SuccessTest(appDO);
         // 测试删除app失败
-        deleteApp2FailureTest();
+        deleteApp2FailureTest(appDO);
     }
 
-    @Rollback()
     private void deleteApp2SuccessTest(AppDO appDO) {
+        appService.addApp(appDO, "admin");
+
         int result = appService.deleteApp(appDO, "admin");
         Assert.assertEquals(result, 1);
     }
 
-    @Rollback()
+    private void deleteApp2FailureTest(AppDO appDO) {
-    private void deleteApp2FailureTest() {
+        appService.addApp(appDO, "admin");
+
         int result = appService.deleteApp(new AppDO(), "admin");
         Assert.assertEquals(result, 0);
     }
@@ -116,7 +118,6 @@ public class AppServiceTest extends BaseTest {
         Assert.assertEquals(result.getCode(), ResultStatus.USER_WITHOUT_AUTHORITY.getCode());
     }
 
-    @Rollback()
     private void updateByAppId2SucessTest(AppDTO appDTO) {
         ResultStatus result1 = appService.updateByAppId(appDTO, "admin", false);
         Assert.assertEquals(result1.getCode(), ResultStatus.SUCCESS.getCode());

kafka-manager-core/src/test/java/com/xiaojukeji/kafka/manager/service/service/gateway/TopicConnectionServiceTest.java

@@ -0,0 +1,85 @@
+package com.xiaojukeji.kafka.manager.service.service.gateway;
+
+import com.xiaojukeji.kafka.manager.common.entity.ao.topic.TopicConnection;
+import com.xiaojukeji.kafka.manager.common.entity.pojo.gateway.TopicConnectionDO;
+import com.xiaojukeji.kafka.manager.service.config.BaseTest;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.testng.Assert;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+/**
+ * @author xuguang
+ * @Date 2021/12/7
+ */
+public class TopicConnectionServiceTest extends BaseTest {
+
+    @Autowired
+    private TopicConnectionService topicConnectionService;
+
+    private static final String TOPIC_NAME = "moduleTest";
+
+    private static final Long CLUSTER_ID = 1L;
+
+    private static final String APP_ID = "dkm_admin";
+
+    @DataProvider(name = "provideTopicConnection")
+    public static Object[][] provideTopicConnection() {
+        TopicConnectionDO topicConnectionDO = new TopicConnectionDO();
+        topicConnectionDO.setId(13L);
+        topicConnectionDO.setAppId(APP_ID);
+        topicConnectionDO.setClusterId(CLUSTER_ID);
+        topicConnectionDO.setTopicName(TOPIC_NAME);
+        topicConnectionDO.setType("fetch");
+        topicConnectionDO.setIp("172.23.142.253");
+        topicConnectionDO.setClientVersion("2.4");
+        topicConnectionDO.setCreateTime(new Date(1638786493173L));
+        return new Object[][] {{topicConnectionDO}};
+    }
+
+    // 测试批量插入为空的情况
+    @Test
+    private void batchAdd2EmptyTest() {
+        topicConnectionService.batchAdd(new ArrayList<>());
+    }
+
+    // 测试批量插入成功的情况，通过调整list的数量和TopicConnectionServiceImpl中splitInterval的数量，使每个流程都测试一遍
+    @Test(dataProvider = "provideTopicConnection")
+    private void batchAdd2SuccessTest(TopicConnectionDO topicConnectionDO) {
+        List<TopicConnectionDO> list = new ArrayList<>();
+        list.add(topicConnectionDO);
+        list.add(topicConnectionDO);
+        list.add(topicConnectionDO);
+        topicConnectionService.batchAdd(list);
+    }
+
+    @Test(dataProvider = "provideTopicConnection")
+    public void getByTopicName2Test(TopicConnectionDO topicConnectionDO) {
+        List<TopicConnection> result = topicConnectionService.getByTopicName(CLUSTER_ID, TOPIC_NAME, new Date(0L), new Date());
+        Assert.assertEquals(result.size(), 1);
+        Assert.assertEquals(result.get(0).toString(), topicConnectionDO.toString());
+    }
+
+    // 测试获取数据时为空
+    @Test
+    public void getByTopicName2EmptyTest() {
+        List<TopicConnection> result = topicConnectionService.getByTopicName(100L, "xgTestxxx", new Date(0L), new Date());
+        Assert.assertTrue(result.isEmpty());
+    }
+
+    // 测试获取数据,clusterId不为null，TODO
+    @Test(dataProvider = "provideTopicConnection")
+    public void getByTopicName2SuccessTest(TopicConnectionDO topicConnectionDO) {
+        List<TopicConnectionDO> list = new ArrayList<>();
+        list.add(topicConnectionDO);
+        topicConnectionService.batchAdd(list);
+        
+        List<TopicConnection> result = topicConnectionService.getByTopicName(CLUSTER_ID, TOPIC_NAME, new Date(0L), new Date());
+        Assert.assertTrue(result.stream().anyMatch(topicConnection -> topicConnection.getTopicName().equals(TOPIC_NAME)
+                                                            && topicConnection.getClusterId().equals(CLUSTER_ID)));
+    }
+}

kafka-manager-core/src/test/java/com/xiaojukeji/kafka/manager/service/utils/TopicCommandsTest.java

@@ -0,0 +1,234 @@
+package com.xiaojukeji.kafka.manager.service.utils;
+
+import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
+import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterDO;
+import com.xiaojukeji.kafka.manager.service.config.BaseTest;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.Properties;
+
+/**
+ * @author xuguang
+ * @Date 2022/1/6
+ */
+public class TopicCommandsTest extends BaseTest {
+
+    private final static Long REAL_CLUSTER_ID_IN_MYSQL = 1L;
+
+    private final static String TEST_CREATE_TOPIC = "createTopicTest";
+
+    private final static String REAL_TOPIC_IN_ZK = "moduleTest";
+
+    private final static String INVALID_TOPIC = ".,&";
+
+    private final static Integer PARTITION_NUM = 1;
+
+    private final static Integer REPLICA_NUM = 1;
+
+    private final static Integer BROKER_ID = 1;
+
+    public ClusterDO getClusterDO() {
+        ClusterDO clusterDO = new ClusterDO();
+        clusterDO.setId(REAL_CLUSTER_ID_IN_MYSQL);
+        clusterDO.setClusterName("LogiKM_moduleTest");
+        clusterDO.setZookeeper("10.190.46.198:2181,10.190.14.237:2181,10.190.50.65:2181/xg");
+        clusterDO.setBootstrapServers("10.190.46.198:9093,10.190.14.237:9093,10.190.50.65:9093");
+        clusterDO.setSecurityProperties("{ \t\"security.protocol\": \"SASL_PLAINTEXT\", \t\"sasl.mechanism\": \"PLAIN\", \t\"sasl.jaas.config\": \"org.apache.kafka.common.security.plain.PlainLoginModule required username=\\\"dkm_admin\\\" password=\\\"km_kMl4N8as1Kp0CCY\\\";\" }");
+        clusterDO.setStatus(1);
+        clusterDO.setGmtCreate(new Date());
+        clusterDO.setGmtModify(new Date());
+        return clusterDO;
+    }
+
+
+    @Test(description = "测试创建topic")
+    public void createTopicTest() {
+        // NullPointerException
+        createTopic2NullPointerExceptionTest();
+        // InvalidPartitions
+        createTopic2InvalidPartitionsTest();
+        // InvalidReplicationFactor
+        createTopic2InvalidReplicationFactorTest();
+        // topic exists
+        createTopic2TopicExistsTest();
+        // invalid topic
+        createTopic2InvalidTopicTest();
+        // success
+        createTopic2SuccessTest();
+    }
+
+    private void createTopic2NullPointerExceptionTest() {
+        ClusterDO clusterDO = getClusterDO();
+        clusterDO.setZookeeper(null);
+        clusterDO.setBootstrapServers(null);
+        ResultStatus result = TopicCommands.createTopic(
+                clusterDO,
+                TEST_CREATE_TOPIC,
+                PARTITION_NUM,
+                REPLICA_NUM,
+                Arrays.asList(BROKER_ID),
+                new Properties()
+        );
+        Assert.assertEquals(result.getCode(), ResultStatus.TOPIC_OPERATION_PARAM_NULL_POINTER.getCode());
+    }
+
+    private void createTopic2InvalidPartitionsTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus result = TopicCommands.createTopic(
+                clusterDO,
+                TEST_CREATE_TOPIC,
+                -1,
+                REPLICA_NUM,
+                Arrays.asList(BROKER_ID),
+                new Properties()
+        );
+        Assert.assertEquals(result.getCode(), ResultStatus.TOPIC_OPERATION_PARTITION_NUM_ILLEGAL.getCode());
+    }
+
+    private void createTopic2InvalidReplicationFactorTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus result = TopicCommands.createTopic(
+                clusterDO,
+                TEST_CREATE_TOPIC,
+                PARTITION_NUM,
+                REPLICA_NUM,
+                Collections.emptyList(),
+                new Properties()
+        );
+        Assert.assertEquals(result.getCode(), ResultStatus.BROKER_NUM_NOT_ENOUGH.getCode());
+    }
+
+    private void createTopic2TopicExistsTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus result = TopicCommands.createTopic(
+                clusterDO,
+                REAL_TOPIC_IN_ZK,
+                PARTITION_NUM,
+                REPLICA_NUM,
+                Arrays.asList(BROKER_ID),
+                new Properties()
+        );
+        Assert.assertEquals(result.getCode(), ResultStatus.TOPIC_OPERATION_TOPIC_EXISTED.getCode());
+    }
+
+    private void createTopic2InvalidTopicTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus result = TopicCommands.createTopic(
+                clusterDO,
+                INVALID_TOPIC,
+                PARTITION_NUM,
+                REPLICA_NUM,
+                Arrays.asList(BROKER_ID),
+                new Properties()
+        );
+        Assert.assertEquals(result.getCode(), ResultStatus.TOPIC_OPERATION_TOPIC_NAME_ILLEGAL.getCode());
+    }
+
+    private void createTopic2SuccessTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus result = TopicCommands.createTopic(
+                clusterDO,
+                TEST_CREATE_TOPIC,
+                PARTITION_NUM,
+                REPLICA_NUM,
+                Arrays.asList(BROKER_ID),
+                new Properties()
+        );
+        Assert.assertEquals(result.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    @Test(description = "测试修改topic配置")
+    public void modifyTopicConfigTest() {
+        // AdminOperationException
+        modifyTopicConfig2AdminOperationExceptionTest();
+        // InvalidConfigurationException
+        modifyTopicConfig2InvalidConfigurationExceptionTest();
+        // success
+        modifyTopicConfig2SuccessTest();
+    }
+
+    private void modifyTopicConfig2AdminOperationExceptionTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus result = TopicCommands.modifyTopicConfig(clusterDO, INVALID_TOPIC, new Properties());
+        Assert.assertEquals(result.getCode(), ResultStatus.TOPIC_OPERATION_UNKNOWN_TOPIC_PARTITION.getCode());
+    }
+
+    private void modifyTopicConfig2InvalidConfigurationExceptionTest() {
+        ClusterDO clusterDO = getClusterDO();
+        Properties properties = new Properties();
+        properties.setProperty("xxx", "xxx");
+        ResultStatus result = TopicCommands.modifyTopicConfig(clusterDO, REAL_TOPIC_IN_ZK, properties);
+        Assert.assertEquals(result.getCode(), ResultStatus.TOPIC_OPERATION_TOPIC_CONFIG_ILLEGAL.getCode());
+    }
+
+    private void modifyTopicConfig2SuccessTest() {
+        ClusterDO clusterDO = getClusterDO();
+        Properties properties = new Properties();
+        ResultStatus result = TopicCommands.modifyTopicConfig(clusterDO, REAL_TOPIC_IN_ZK, properties);
+        Assert.assertEquals(result.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    @Test(description = "测试扩分区")
+    public void expandTopicTest() {
+        // failed
+        expandTopic2FailureTest();
+        // success
+        expandTopic2SuccessTest();
+    }
+
+    private void expandTopic2FailureTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus result = TopicCommands.expandTopic(
+                clusterDO,
+                INVALID_TOPIC,
+                PARTITION_NUM + 1,
+                Arrays.asList(BROKER_ID, 2)
+        );
+        Assert.assertEquals(result.getCode(), ResultStatus.TOPIC_OPERATION_UNKNOWN_ERROR.getCode());
+    }
+
+    private void expandTopic2SuccessTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus result = TopicCommands.expandTopic(
+                clusterDO,
+                TEST_CREATE_TOPIC,
+                PARTITION_NUM + 1,
+                Arrays.asList(BROKER_ID, 2)
+        );
+        Assert.assertEquals(result.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    @Test(description = "测试删除Topic")
+    public void deleteTopicTest() {
+        // UnknownTopicOrPartitionException
+        deleteTopic2UnknownTopicOrPartitionExceptionTest();
+        // NullPointerException
+        deleteTopic2NullPointerExceptionTest();
+        // success
+        deleteTopic2SuccessTest();
+    }
+
+    private void deleteTopic2UnknownTopicOrPartitionExceptionTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus result = TopicCommands.deleteTopic(clusterDO, INVALID_TOPIC);
+        Assert.assertEquals(result.getCode(), ResultStatus.TOPIC_OPERATION_UNKNOWN_TOPIC_PARTITION.getCode());
+    }
+
+    private void deleteTopic2NullPointerExceptionTest() {
+        ClusterDO clusterDO = getClusterDO();
+        clusterDO.setBootstrapServers(null);
+        clusterDO.setZookeeper(null);
+        ResultStatus result = TopicCommands.deleteTopic(clusterDO, INVALID_TOPIC);
+        Assert.assertEquals(result.getCode(), ResultStatus.TOPIC_OPERATION_UNKNOWN_ERROR.getCode());
+    }
+
+    private void deleteTopic2SuccessTest() {
+        ClusterDO clusterDO = getClusterDO();
+        ResultStatus result = TopicCommands.deleteTopic(clusterDO, TEST_CREATE_TOPIC);
+        Assert.assertEquals(result.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+}

kafka-manager-core/src/test/java/com/xiaojukeji/kafka/manager/service/utils/TopicReassignUtilsTest.java

@@ -0,0 +1,78 @@
+package com.xiaojukeji.kafka.manager.service.utils;
+
+import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterDO;
+import com.xiaojukeji.kafka.manager.service.config.BaseTest;
+import org.junit.Assert;
+import org.testng.annotations.Test;
+
+import java.util.Arrays;
+import java.util.Date;
+
+/**
+ * @author xuguang
+ * @Date 2022/1/6
+ */
+public class TopicReassignUtilsTest extends BaseTest {
+
+    private final static Long REAL_CLUSTER_ID_IN_MYSQL = 1L;
+
+    private final static String TEST_CREATE_TOPIC = "createTopicTest";
+
+    private final static String REAL_TOPIC_IN_ZK = "moduleTest";
+
+    private final static String INVALID_TOPIC = ".,&";
+
+    private final static Integer PARTITION_NUM = 1;
+
+    private final static Integer REPLICA_NUM = 1;
+
+    private final static Integer BROKER_ID = 1;
+
+    private final static Integer PARTITION_ID = 1;
+
+    public ClusterDO getClusterDO() {
+        ClusterDO clusterDO = new ClusterDO();
+        clusterDO.setId(REAL_CLUSTER_ID_IN_MYSQL);
+        clusterDO.setClusterName("LogiKM_moduleTest");
+        clusterDO.setZookeeper("10.190.46.198:2181,10.190.14.237:2181,10.190.50.65:2181/xg");
+        clusterDO.setBootstrapServers("10.190.46.198:9093,10.190.14.237:9093,10.190.50.65:9093");
+        clusterDO.setSecurityProperties("{ \t\"security.protocol\": \"SASL_PLAINTEXT\", \t\"sasl.mechanism\": \"PLAIN\", \t\"sasl.jaas.config\": \"org.apache.kafka.common.security.plain.PlainLoginModule required username=\\\"dkm_admin\\\" password=\\\"km_kMl4N8as1Kp0CCY\\\";\" }");
+        clusterDO.setStatus(1);
+        clusterDO.setGmtCreate(new Date());
+        clusterDO.setGmtModify(new Date());
+        return clusterDO;
+    }
+
+    @Test
+    public void generateReassignmentJsonTest() {
+        // null
+        generateReassignmentJson2NullPointerExceptionTest();
+        // not null
+        generateReassignmentJson2SuccessTest();
+    }
+
+    private void generateReassignmentJson2NullPointerExceptionTest() {
+        ClusterDO clusterDO = getClusterDO();
+        clusterDO.setZookeeper(null);
+        String result = TopicReassignUtils.generateReassignmentJson(
+                clusterDO,
+                REAL_TOPIC_IN_ZK,
+                Arrays.asList(PARTITION_ID),
+                Arrays.asList(BROKER_ID)
+        );
+
+        Assert.assertNull(result);
+    }
+
+    private void generateReassignmentJson2SuccessTest() {
+        ClusterDO clusterDO = getClusterDO();
+        String result = TopicReassignUtils.generateReassignmentJson(
+                clusterDO,
+                REAL_TOPIC_IN_ZK,
+                Arrays.asList(PARTITION_ID),
+                Arrays.asList(BROKER_ID)
+        );
+
+        Assert.assertNotNull(result);
+    }
+}

kafka-manager-extends/kafka-manager-kcm/src/test/java/com/xiaojukeji/kafka/manager/kcm/ClusterHostTaskServiceTest.java

@@ -0,0 +1,80 @@
+package com.xiaojukeji.kafka.manager.kcm;
+
+import com.xiaojukeji.kafka.manager.common.entity.Result;
+import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
+import com.xiaojukeji.kafka.manager.kcm.common.entry.ao.CreationTaskData;
+import com.xiaojukeji.kafka.manager.kcm.common.entry.dto.ClusterHostTaskDTO;
+import com.xiaojukeji.kafka.manager.kcm.config.BaseTest;
+import com.xiaojukeji.kafka.manager.kcm.tasks.ClusterHostTaskService;
+import org.junit.Assert;
+import org.mockito.InjectMocks;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.testng.annotations.Test;
+
+import java.util.Arrays;
+
+/**
+ * @author xuguang
+ * @Date 2022/1/5
+ */
+public class ClusterHostTaskServiceTest extends BaseTest {
+
+    @Autowired
+    @InjectMocks
+    private ClusterHostTaskService clusterHostTaskService;
+
+    private ClusterHostTaskDTO getClusterHostTaskDTO() {
+        ClusterHostTaskDTO clusterHostTaskDTO = new ClusterHostTaskDTO();
+        clusterHostTaskDTO.setClusterId(-1L);
+        clusterHostTaskDTO.setHostList(Arrays.asList("127.0.0.1"));
+        clusterHostTaskDTO.setTaskType("");
+        clusterHostTaskDTO.setKafkaFileBaseUrl("");
+        clusterHostTaskDTO.setKafkaPackageMd5("");
+        clusterHostTaskDTO.setKafkaPackageName("");
+        clusterHostTaskDTO.setServerPropertiesMd5("");
+        clusterHostTaskDTO.setServerPropertiesName("");
+        return clusterHostTaskDTO;
+    }
+
+    @Test
+    public void getOperationHostsTest() {
+        // cluster task host list illegal
+        getOperationHosts2HostListIllegalTest();
+        // success
+        getOperationHosts2SuccessTest();
+    }
+
+    private void getOperationHosts2HostListIllegalTest() {
+        ClusterHostTaskDTO clusterHostTaskDTO = getClusterHostTaskDTO();
+        clusterHostTaskDTO.setHostList(Arrays.asList(""));
+        Result<CreationTaskData> result = clusterHostTaskService.getOperationHosts(clusterHostTaskDTO);
+        Assert.assertEquals(result.getCode(), ResultStatus.CLUSTER_TASK_HOST_LIST_ILLEGAL.getCode());
+    }
+
+    private void getOperationHosts2SuccessTest() {
+        ClusterHostTaskDTO clusterHostTaskDTO = getClusterHostTaskDTO();
+        Result<CreationTaskData> result = clusterHostTaskService.getOperationHosts(clusterHostTaskDTO);
+        Assert.assertEquals(result.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    @Test
+    public void getCreateTaskParamDTOTest() {
+        // not success
+        getCreateTaskParamDTO2NotSuccessTest();
+        // success
+        getCreateTaskParamDTO2SuccessTest();
+    }
+
+    private void getCreateTaskParamDTO2NotSuccessTest() {
+        ClusterHostTaskDTO clusterHostTaskDTO = getClusterHostTaskDTO();
+        clusterHostTaskDTO.setHostList(Arrays.asList(""));
+        Result<CreationTaskData> dto = clusterHostTaskService.getCreateTaskParamDTO(clusterHostTaskDTO);
+        Assert.assertEquals(dto.getCode(), ResultStatus.CLUSTER_TASK_HOST_LIST_ILLEGAL.getCode());
+    }
+
+    private void getCreateTaskParamDTO2SuccessTest() {
+        ClusterHostTaskDTO clusterHostTaskDTO = getClusterHostTaskDTO();
+        Result<CreationTaskData> dto = clusterHostTaskService.getCreateTaskParamDTO(clusterHostTaskDTO);
+        Assert.assertEquals(dto.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+}

kafka-manager-extends/kafka-manager-kcm/src/test/java/com/xiaojukeji/kafka/manager/kcm/ClusterRoleTaskServiceTest.java

@@ -0,0 +1,74 @@
+package com.xiaojukeji.kafka.manager.kcm;
+
+import com.xiaojukeji.kafka.manager.common.bizenum.KafkaBrokerRoleEnum;
+import com.xiaojukeji.kafka.manager.common.entity.Result;
+import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
+import com.xiaojukeji.kafka.manager.kcm.common.entry.ao.CreationTaskData;
+import com.xiaojukeji.kafka.manager.kcm.common.entry.dto.ClusterRoleTaskDTO;
+import com.xiaojukeji.kafka.manager.kcm.config.BaseTest;
+import com.xiaojukeji.kafka.manager.kcm.tasks.ClusterRoleTaskService;
+import org.junit.Assert;
+import org.mockito.InjectMocks;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.testng.annotations.Test;
+
+import java.util.*;
+
+/**
+ * @author xuguang
+ * @Date 2022/1/5
+ */
+public class ClusterRoleTaskServiceTest extends BaseTest {
+
+    @Autowired
+    @InjectMocks
+    private ClusterRoleTaskService clusterRoleTaskService;
+
+    private ClusterRoleTaskDTO getClusterRoleTaskDTO() {
+        ClusterRoleTaskDTO clusterRoleTaskDTO = new ClusterRoleTaskDTO();
+        clusterRoleTaskDTO.setClusterId(-1L);
+        clusterRoleTaskDTO.setTaskType("");
+        return clusterRoleTaskDTO;
+    }
+
+    @Test
+    public void getOperationHostsTest() {
+        // controller not alive
+        getOperationHosts2ControllerNotAliveTest();
+        // success
+        getOperationHosts2SuccessTest();
+    }
+
+    private void getOperationHosts2ControllerNotAliveTest() {
+        ClusterRoleTaskDTO dto = getClusterRoleTaskDTO();
+        List<String> upgradeSequenceList = new ArrayList<>();
+        upgradeSequenceList.add(KafkaBrokerRoleEnum.CONTROLLER.getRole());
+        dto.setUpgradeSequenceList(upgradeSequenceList);
+        dto.setKafkaRoleBrokerHostMap(new HashMap<>(0));
+
+        Result<CreationTaskData> result = clusterRoleTaskService.getOperationHosts(dto);
+        Assert.assertEquals(result.getCode(), ResultStatus.CONTROLLER_NOT_ALIVE.getCode());
+    }
+
+    private void getOperationHosts2SuccessTest() {
+        ClusterRoleTaskDTO dto = getClusterRoleTaskDTO();
+        List<String> upgradeSequenceList = new ArrayList<>();
+        upgradeSequenceList.add(KafkaBrokerRoleEnum.CONTROLLER.getRole());
+        upgradeSequenceList.add(KafkaBrokerRoleEnum.NORMAL.getRole());
+        upgradeSequenceList.add(KafkaBrokerRoleEnum.COORDINATOR.getRole());
+        dto.setUpgradeSequenceList(upgradeSequenceList);
+        Map<String, List<String>> map = new HashMap<>();
+        List<String> controllerList = new ArrayList<>();
+        controllerList.add("127.0.0.1");
+        controllerList.add("localhost");
+        List<String> coordinatorList = new ArrayList<>();
+        coordinatorList.add("127.0.0.1");
+        coordinatorList.add("localhost");
+        map.put(KafkaBrokerRoleEnum.CONTROLLER.getRole(), controllerList);
+        map.put(KafkaBrokerRoleEnum.COORDINATOR.getRole(), coordinatorList);
+        dto.setKafkaRoleBrokerHostMap(map);
+
+        Result<CreationTaskData> result = clusterRoleTaskService.getOperationHosts(dto);
+        Assert.assertEquals(result.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+}

kafka-manager-extends/kafka-manager-kcm/src/test/java/com/xiaojukeji/kafka/manager/kcm/ClusterTaskServiceTest.java

@@ -1,8 +1,354 @@
 package com.xiaojukeji.kafka.manager.kcm;
 
+import com.xiaojukeji.kafka.manager.common.entity.Result;
+import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
+import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterTaskDO;
+import com.xiaojukeji.kafka.manager.dao.ClusterTaskDao;
+import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskActionEnum;
+import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskStateEnum;
+import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskTypeEnum;
+import com.xiaojukeji.kafka.manager.kcm.common.entry.ao.ClusterTaskLog;
+import com.xiaojukeji.kafka.manager.kcm.common.entry.ao.ClusterTaskStatus;
+import com.xiaojukeji.kafka.manager.kcm.common.entry.dto.AbstractClusterTaskDTO;
+import com.xiaojukeji.kafka.manager.kcm.common.entry.dto.ClusterHostTaskDTO;
+import com.xiaojukeji.kafka.manager.kcm.component.agent.AbstractAgent;
+import com.xiaojukeji.kafka.manager.kcm.config.BaseTest;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.util.Arrays;
+
 /**
  * @author xuguang
  * @Date 2021/12/27
  */
-public class ClusterTaskServiceTest {
+public class ClusterTaskServiceTest extends BaseTest {
+
+    private static final Long REAL_CLUSTER_ID_IN_MYSQL = 1L;
+
+    private static final String ADMIN = "admin";
+
+    private static final String BASEURL = "127.0.0.1";
+
+    private static final String MD5 = "md5";
+
+    private static final Long REAL_TASK_ID_IN_MYSQL = 1L;
+
+    private static final Long INVALID_TASK_ID = -1L;
+
+
+    @Autowired
+    @InjectMocks
+    private ClusterTaskService clusterTaskService;
+
+    @Mock
+    private AbstractAgent abstractAgent;
+
+    @Mock
+    private ClusterTaskDao clusterTaskDao;
+
+    private ClusterHostTaskDTO getClusterHostTaskDTO() {
+        ClusterHostTaskDTO clusterHostTaskDTO = new ClusterHostTaskDTO();
+        clusterHostTaskDTO.setClusterId(-1L);
+        clusterHostTaskDTO.setHostList(Arrays.asList(BASEURL));
+        clusterHostTaskDTO.setTaskType(ClusterTaskTypeEnum.HOST_UPGRADE.getName());
+        clusterHostTaskDTO.setKafkaFileBaseUrl(BASEURL);
+        clusterHostTaskDTO.setKafkaPackageMd5(MD5);
+        clusterHostTaskDTO.setKafkaPackageName("name");
+        clusterHostTaskDTO.setServerPropertiesMd5(MD5);
+        clusterHostTaskDTO.setServerPropertiesName("name");
+        return clusterHostTaskDTO;
+    }
+
+    private ClusterTaskDO getClusterTaskDO() {
+        ClusterTaskDO clusterTaskDO = new ClusterTaskDO();
+        clusterTaskDO.setClusterId(REAL_CLUSTER_ID_IN_MYSQL);
+        clusterTaskDO.setId(REAL_TASK_ID_IN_MYSQL);
+        clusterTaskDO.setAgentTaskId(-1L);
+        clusterTaskDO.setAgentRollbackTaskId(-1L);
+        clusterTaskDO.setTaskStatus(0);
+        return clusterTaskDO;
+    }
+
+    @BeforeMethod
+    public void setup() {
+        MockitoAnnotations.initMocks(this);
+    }
+
+    @Test(description = "测试创建任务")
+    public void createTaskTest() {
+        // paramIllegal
+        createTask2ParamIllegalTest();
+        // not success
+        createTask2NotSuccessTest();
+        // CallClusterTaskAgentFailed
+        createTask2CallClusterTaskAgentFailedTest();
+        // success
+        createTask2SuccessTest();
+        // mysqlError
+        createTask2MysqlErrorTest();
+    }
+
+    private void createTask2ParamIllegalTest() {
+        AbstractClusterTaskDTO dto = getClusterHostTaskDTO();
+        dto.setTaskType(null);
+        Result result1 = clusterTaskService.createTask(dto, ADMIN);
+        Assert.assertEquals(result1.getCode(), ResultStatus.PARAM_ILLEGAL.getCode());
+
+        dto.setTaskType("");
+        Result result2 = clusterTaskService.createTask(dto, ADMIN);
+        Assert.assertEquals(result2.getCode(), ResultStatus.PARAM_ILLEGAL.getCode());
+    }
+
+    private void createTask2NotSuccessTest() {
+        ClusterHostTaskDTO clusterHostTaskDTO = getClusterHostTaskDTO();
+        clusterHostTaskDTO.setHostList(Arrays.asList("host"));
+        Result result = clusterTaskService.createTask(clusterHostTaskDTO, ADMIN);
+        Assert.assertEquals(result.getCode(), ResultStatus.CLUSTER_TASK_HOST_LIST_ILLEGAL.getCode());
+    }
+
+    private void createTask2CallClusterTaskAgentFailedTest() {
+        Mockito.when(abstractAgent.createTask(Mockito.any())).thenReturn(null);
+
+        ClusterHostTaskDTO clusterHostTaskDTO = getClusterHostTaskDTO();
+        Result result = clusterTaskService.createTask(clusterHostTaskDTO, ADMIN);
+        Assert.assertEquals(result.getCode(), ResultStatus.CALL_CLUSTER_TASK_AGENT_FAILED.getCode());
+    }
+
+    private void createTask2SuccessTest() {
+        Mockito.when(abstractAgent.createTask(Mockito.any())).thenReturn(Result.buildSuc(1L));
+        Mockito.when(clusterTaskDao.insert(Mockito.any())).thenReturn(1);
+
+        ClusterHostTaskDTO clusterHostTaskDTO = getClusterHostTaskDTO();
+        Result result = clusterTaskService.createTask(clusterHostTaskDTO, ADMIN);
+        Assert.assertEquals(result.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    private void createTask2MysqlErrorTest() {
+        Mockito.when(abstractAgent.createTask(Mockito.any())).thenReturn(Result.buildSuc(1L));
+        Mockito.when(clusterTaskDao.insert(Mockito.any())).thenThrow(RuntimeException.class);
+
+        ClusterHostTaskDTO clusterHostTaskDTO = getClusterHostTaskDTO();
+        Result result = clusterTaskService.createTask(clusterHostTaskDTO, ADMIN);
+        Assert.assertEquals(result.getCode(), ResultStatus.MYSQL_ERROR.getCode());
+    }
+
+    @Test
+    public void executeTaskTest() {
+        // task not exist
+        executeTask2TaskNotExistTest();
+        // CallClusterTaskAgentFailed
+        executeTask2CallClusterTaskAgentFailedTest();
+        // 暂停状态, 可以执行开始
+        executeTask2StartTest();
+        // 运行状态, 可以执行暂停
+        executeTask2PauseTest();
+        // 忽略
+        executeTask2IgnoreTest();
+        // 取消
+        executeTask2CancelTest();
+        // operation failed
+        executeTask2OperationFailedTest();
+        // 回滚 operation forbidden
+        executeTask2RollbackForbiddenTest();
+    }
+
+    private void executeTask2TaskNotExistTest() {
+        ResultStatus resultStatus = clusterTaskService.executeTask(INVALID_TASK_ID, ClusterTaskActionEnum.START.getAction(), ADMIN);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.RESOURCE_NOT_EXIST.getCode());
+    }
+
+    private void executeTask2CallClusterTaskAgentFailedTest() {
+        Mockito.when(abstractAgent.getTaskExecuteState(Mockito.anyLong())).thenReturn(null);
+        ClusterTaskDO clusterTaskDO = getClusterTaskDO();
+        Mockito.when(clusterTaskDao.getById(Mockito.anyLong())).thenReturn(clusterTaskDO);
+
+        ResultStatus resultStatus = clusterTaskService.executeTask(REAL_TASK_ID_IN_MYSQL, ClusterTaskActionEnum.START.getAction(), ADMIN);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.CALL_CLUSTER_TASK_AGENT_FAILED.getCode());
+    }
+
+    private void executeTask2StartTest() {
+        Mockito.when(abstractAgent.getTaskExecuteState(Mockito.anyLong())).thenReturn(Result.buildSuc(ClusterTaskStateEnum.BLOCKED));
+        ClusterTaskDO clusterTaskDO = getClusterTaskDO();
+        Mockito.when(clusterTaskDao.getById(Mockito.anyLong())).thenReturn(clusterTaskDO);
+
+        // success
+        Mockito.when(abstractAgent.actionTask(Mockito.anyLong(), Mockito.any())).thenReturn(true);
+        ResultStatus resultStatus = clusterTaskService.executeTask(REAL_TASK_ID_IN_MYSQL, ClusterTaskActionEnum.START.getAction(), ADMIN);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.SUCCESS.getCode());
+
+        // operation failed
+        Mockito.when(abstractAgent.actionTask(Mockito.anyLong(), Mockito.any())).thenReturn(false);
+        ResultStatus resultStatus2 = clusterTaskService.executeTask(REAL_TASK_ID_IN_MYSQL, ClusterTaskActionEnum.START.getAction(), ADMIN);
+        Assert.assertEquals(resultStatus2.getCode(), ResultStatus.OPERATION_FAILED.getCode());
+    }
+
+    private void executeTask2PauseTest() {
+        Mockito.when(abstractAgent.getTaskExecuteState(Mockito.anyLong())).thenReturn(Result.buildSuc(ClusterTaskStateEnum.RUNNING));
+        ClusterTaskDO clusterTaskDO = getClusterTaskDO();
+        Mockito.when(clusterTaskDao.getById(Mockito.anyLong())).thenReturn(clusterTaskDO);
+
+        // success
+        Mockito.when(abstractAgent.actionTask(Mockito.anyLong(), Mockito.any())).thenReturn(true);
+        ResultStatus resultStatus = clusterTaskService.executeTask(REAL_TASK_ID_IN_MYSQL, ClusterTaskActionEnum.PAUSE.getAction(), ADMIN);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.SUCCESS.getCode());
+
+        // operation failed
+        Mockito.when(abstractAgent.actionTask(Mockito.anyLong(), Mockito.any())).thenReturn(false);
+        ResultStatus resultStatus2 = clusterTaskService.executeTask(REAL_TASK_ID_IN_MYSQL, ClusterTaskActionEnum.PAUSE.getAction(), ADMIN);
+        Assert.assertEquals(resultStatus2.getCode(), ResultStatus.OPERATION_FAILED.getCode());
+    }
+
+    private void executeTask2IgnoreTest() {
+        Mockito.when(abstractAgent.getTaskExecuteState(Mockito.anyLong())).thenReturn(Result.buildSuc(ClusterTaskStateEnum.RUNNING));
+        ClusterTaskDO clusterTaskDO = getClusterTaskDO();
+        Mockito.when(clusterTaskDao.getById(Mockito.anyLong())).thenReturn(clusterTaskDO);
+
+        // success
+        Mockito.when(abstractAgent.actionTask(Mockito.anyLong(), Mockito.any())).thenReturn(true);
+        ResultStatus resultStatus = clusterTaskService.executeTask(REAL_TASK_ID_IN_MYSQL, ClusterTaskActionEnum.IGNORE.getAction(), "");
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.SUCCESS.getCode());
+
+        // operation failed
+        Mockito.when(abstractAgent.actionTask(Mockito.anyLong(), Mockito.any())).thenReturn(false);
+        ResultStatus resultStatus2 = clusterTaskService.executeTask(REAL_TASK_ID_IN_MYSQL, ClusterTaskActionEnum.IGNORE.getAction(), "");
+        Assert.assertEquals(resultStatus2.getCode(), ResultStatus.OPERATION_FAILED.getCode());
+    }
+
+    private void executeTask2CancelTest() {
+        Mockito.when(abstractAgent.getTaskExecuteState(Mockito.anyLong())).thenReturn(Result.buildSuc(ClusterTaskStateEnum.RUNNING));
+        ClusterTaskDO clusterTaskDO = getClusterTaskDO();
+        Mockito.when(clusterTaskDao.getById(Mockito.anyLong())).thenReturn(clusterTaskDO);
+
+        // success
+        Mockito.when(abstractAgent.actionTask(Mockito.anyLong(), Mockito.any())).thenReturn(true);
+        ResultStatus resultStatus = clusterTaskService.executeTask(REAL_TASK_ID_IN_MYSQL, ClusterTaskActionEnum.CANCEL.getAction(), "");
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.SUCCESS.getCode());
+
+        // operation failed
+        Mockito.when(abstractAgent.actionTask(Mockito.anyLong(), Mockito.any())).thenReturn(false);
+        ResultStatus resultStatus2 = clusterTaskService.executeTask(REAL_TASK_ID_IN_MYSQL, ClusterTaskActionEnum.CANCEL.getAction(), "");
+        Assert.assertEquals(resultStatus2.getCode(), ResultStatus.OPERATION_FAILED.getCode());
+    }
+
+    private void executeTask2OperationFailedTest() {
+        Mockito.when(abstractAgent.getTaskExecuteState(Mockito.anyLong())).thenReturn(Result.buildSuc(ClusterTaskStateEnum.RUNNING));
+        ClusterTaskDO clusterTaskDO = getClusterTaskDO();
+        Mockito.when(clusterTaskDao.getById(Mockito.anyLong())).thenReturn(clusterTaskDO);
+
+        // operation failed
+        ResultStatus resultStatus2 = clusterTaskService.executeTask(REAL_TASK_ID_IN_MYSQL, ClusterTaskActionEnum.START.getAction(), ADMIN);
+        Assert.assertEquals(resultStatus2.getCode(), ResultStatus.OPERATION_FAILED.getCode());
+    }
+
+    private void executeTask2RollbackForbiddenTest() {
+        Mockito.when(abstractAgent.getTaskExecuteState(Mockito.anyLong())).thenReturn(Result.buildSuc(ClusterTaskStateEnum.RUNNING));
+        ClusterTaskDO clusterTaskDO = getClusterTaskDO();
+        Mockito.when(clusterTaskDao.getById(Mockito.anyLong())).thenReturn(clusterTaskDO);
+
+        // operation failed
+        ResultStatus resultStatus2 = clusterTaskService.executeTask(REAL_TASK_ID_IN_MYSQL, ClusterTaskActionEnum.ROLLBACK.getAction(), ADMIN);
+        Assert.assertEquals(resultStatus2.getCode(), ResultStatus.OPERATION_FORBIDDEN.getCode());
+    }
+
+    @Test
+    public void getTaskLogTest() {
+        // task not exist
+        getTaskLog2TaskNotExistTest();
+        // call cluster task agent failed
+        getTaskLog2CallClusterTaskAgentFailedTest();
+        // success
+        getTaskLog2SuccessTest();
+    }
+
+    private void getTaskLog2TaskNotExistTest() {
+        Mockito.when(clusterTaskDao.getById(Mockito.anyLong())).thenReturn(null);
+
+        Result<String> result = clusterTaskService.getTaskLog(REAL_TASK_ID_IN_MYSQL, ADMIN);
+        Assert.assertEquals(result.getCode(), ResultStatus.TASK_NOT_EXIST.getCode());
+    }
+
+    private void getTaskLog2CallClusterTaskAgentFailedTest() {
+        ClusterTaskDO clusterTaskDO = getClusterTaskDO();
+        Mockito.when(clusterTaskDao.getById(Mockito.anyLong())).thenReturn(clusterTaskDO);
+        Mockito.when(abstractAgent.getTaskLog(Mockito.anyLong(), Mockito.any())).thenReturn(null);
+
+        Result<String> result = clusterTaskService.getTaskLog(REAL_TASK_ID_IN_MYSQL, ADMIN);
+        Assert.assertEquals(result.getCode(), ResultStatus.CALL_CLUSTER_TASK_AGENT_FAILED.getCode());
+    }
+
+    private void getTaskLog2SuccessTest() {
+        ClusterTaskDO clusterTaskDO = getClusterTaskDO();
+        Mockito.when(clusterTaskDao.getById(Mockito.anyLong())).thenReturn(clusterTaskDO);
+        Mockito.when(abstractAgent.getTaskLog(Mockito.anyLong(), Mockito.any()))
+                .thenReturn(Result.buildSuc(new ClusterTaskLog("")));
+
+        Result<String> result = clusterTaskService.getTaskLog(REAL_TASK_ID_IN_MYSQL, ADMIN);
+        Assert.assertEquals(result.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    @Test
+    public void getTaskStateTest() {
+        // null
+        getTaskState2NullTest();
+        //
+        getTaskState2NotNullTest();
+    }
+
+    private void getTaskState2NullTest() {
+        Mockito.when(abstractAgent.getTaskExecuteState(Mockito.anyLong())).thenReturn(null);
+
+        ClusterTaskStateEnum taskState = clusterTaskService.getTaskState(REAL_TASK_ID_IN_MYSQL);
+        Assert.assertNull(taskState);
+    }
+
+    private void getTaskState2NotNullTest() {
+
+        Mockito.when(abstractAgent.getTaskExecuteState(Mockito.anyLong())).thenReturn(Result.buildSuc(ClusterTaskStateEnum.RUNNING));
+
+        ClusterTaskStateEnum taskState = clusterTaskService.getTaskState(REAL_TASK_ID_IN_MYSQL);
+        Assert.assertNotNull(taskState);
+    }
+
+    @Test
+    public void getTaskStatusTest() {
+        // task not exist
+        getTaskStatus2TaskNotExistTest();
+        // get failed
+        getTaskStatus2FailedTest();
+        // success
+        getTaskStatus2SuccessTest();
+    }
+
+    private void getTaskStatus2TaskNotExistTest() {
+        Mockito.when(clusterTaskDao.getById(Mockito.anyLong())).thenReturn(null);
+
+        Result<ClusterTaskStatus> result = clusterTaskService.getTaskStatus(REAL_TASK_ID_IN_MYSQL);
+        Assert.assertEquals(result.getCode(), ResultStatus.TASK_NOT_EXIST.getCode());
+    }
+
+    private void getTaskStatus2FailedTest() {
+        ClusterTaskDO clusterTaskDO = getClusterTaskDO();
+        Mockito.when(clusterTaskDao.getById(Mockito.anyLong())).thenReturn(clusterTaskDO);
+        Mockito.when(abstractAgent.getTaskExecuteState(Mockito.anyLong())).thenReturn(Result.buildFailure(""));
+
+        Result<ClusterTaskStatus> result = clusterTaskService.getTaskStatus(REAL_TASK_ID_IN_MYSQL);
+        Assert.assertEquals(result.getCode(), ResultStatus.FAIL.getCode());
+    }
+
+    private void getTaskStatus2SuccessTest() {
+        ClusterTaskDO clusterTaskDO = getClusterTaskDO();
+        Mockito.when(clusterTaskDao.getById(Mockito.anyLong())).thenReturn(clusterTaskDO);
+        Mockito.when(abstractAgent.getTaskExecuteState(Mockito.anyLong()))
+                .thenReturn(Result.buildSuc(ClusterTaskStateEnum.RUNNING));
+
+        Result<ClusterTaskStatus> result = clusterTaskService.getTaskStatus(REAL_TASK_ID_IN_MYSQL);
+        Assert.assertEquals(result.getCode(), ResultStatus.SUCCESS.getCode());
+    }
 }

kafka-manager-extends/kafka-manager-kcm/src/test/java/com/xiaojukeji/kafka/manager/kcm/KafkaFileServiceTest.java

@@ -0,0 +1,224 @@
+package com.xiaojukeji.kafka.manager.kcm;
+
+import com.xiaojukeji.kafka.manager.common.bizenum.KafkaFileEnum;
+import com.xiaojukeji.kafka.manager.common.entity.Result;
+import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
+import com.xiaojukeji.kafka.manager.common.entity.dto.normal.KafkaFileDTO;
+import com.xiaojukeji.kafka.manager.common.entity.pojo.KafkaFileDO;
+import com.xiaojukeji.kafka.manager.dao.KafkaFileDao;
+import com.xiaojukeji.kafka.manager.kcm.component.storage.AbstractStorageService;
+import com.xiaojukeji.kafka.manager.kcm.config.BaseTest;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.mock.web.MockMultipartFile;
+import org.springframework.web.multipart.MultipartFile;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * @author xuguang
+ * @Date 2022/1/4
+ */
+public class KafkaFileServiceTest extends BaseTest {
+
+    private static final Long KAFKA_FILE_ID = 1L;
+
+    private static final String ADMIN = "admin";
+
+    private KafkaFileDTO getKafkaFileDTO() {
+        KafkaFileDTO kafkaFileDTO = new KafkaFileDTO();
+        kafkaFileDTO.setId(KAFKA_FILE_ID);
+        kafkaFileDTO.setClusterId(-1L);
+        kafkaFileDTO.setFileMd5("");
+        kafkaFileDTO.setFileName(".tgz");
+        kafkaFileDTO.setFileType(KafkaFileEnum.PACKAGE.getCode());
+        kafkaFileDTO.setUploadFile(new MockMultipartFile("name", new byte[]{1}));
+        return kafkaFileDTO;
+    }
+
+    private KafkaFileDO getKafkaFileDO() {
+        KafkaFileDO kafkaFileDO = new KafkaFileDO();
+        kafkaFileDO.setFileType(KafkaFileEnum.PACKAGE.getCode());
+        kafkaFileDO.setFileName(".tgz");
+        return kafkaFileDO;
+    }
+
+    @Autowired
+    @InjectMocks
+    private KafkaFileService kafkaFileService;
+
+    @Mock
+    private KafkaFileDao kafkaFileDao;
+
+    @Mock
+    private AbstractStorageService storageService;
+
+    @BeforeMethod
+    public void setup() {
+        MockitoAnnotations.initMocks(this);
+    }
+
+    @Test
+    public void uploadKafkaFile() {
+        // paramIllegal
+        uploadKafkaFile2ParamIllegalTest();
+        // mysqlError
+        uploadKafkaFile2MysqlErrorTest();
+        // storage upload file failed
+        uploadKafkaFile2StorageUploadFileFailedTest();
+        // success
+        uploadKafkaFile2SuccessTest();
+        // DuplicateKey
+        uploadKafkaFile2DuplicateKeyTest();
+    }
+
+    private void uploadKafkaFile2ParamIllegalTest() {
+        KafkaFileDTO kafkaFileDTO = getKafkaFileDTO();
+        kafkaFileDTO.setUploadFile(null);
+        ResultStatus resultStatus = kafkaFileService.uploadKafkaFile(kafkaFileDTO, ADMIN);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.PARAM_ILLEGAL.getCode());
+    }
+
+    private void uploadKafkaFile2MysqlErrorTest() {
+        Mockito.when(kafkaFileDao.insert(Mockito.any())).thenReturn(-1);
+
+        KafkaFileDTO kafkaFileDTO = getKafkaFileDTO();
+        ResultStatus resultStatus = kafkaFileService.uploadKafkaFile(kafkaFileDTO, ADMIN);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.MYSQL_ERROR.getCode());
+    }
+
+    private void uploadKafkaFile2StorageUploadFileFailedTest() {
+        Mockito.when(kafkaFileDao.insert(Mockito.any())).thenReturn(1);
+        Mockito.when(kafkaFileDao.deleteById(Mockito.any())).thenReturn(1);
+        Mockito.when(storageService.upload(Mockito.any(), Mockito.any(), Mockito.any())).thenReturn(false);
+
+        KafkaFileDTO kafkaFileDTO = getKafkaFileDTO();
+        ResultStatus resultStatus = kafkaFileService.uploadKafkaFile(kafkaFileDTO, ADMIN);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.STORAGE_UPLOAD_FILE_FAILED.getCode());
+    }
+
+    private void uploadKafkaFile2SuccessTest() {
+        Mockito.when(kafkaFileDao.insert(Mockito.any())).thenReturn(1);
+        Mockito.when(kafkaFileDao.deleteById(Mockito.any())).thenReturn(1);
+        Mockito.when(storageService.upload(Mockito.any(), Mockito.any(), Mockito.any())).thenReturn(true);
+
+        KafkaFileDTO kafkaFileDTO = getKafkaFileDTO();
+        ResultStatus resultStatus = kafkaFileService.uploadKafkaFile(kafkaFileDTO, ADMIN);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    private void uploadKafkaFile2DuplicateKeyTest() {
+        Mockito.when(kafkaFileDao.insert(Mockito.any())).thenThrow(DuplicateKeyException.class);
+        Mockito.when(kafkaFileDao.deleteById(Mockito.any())).thenReturn(1);
+        Mockito.when(storageService.upload(Mockito.any(), Mockito.any(), Mockito.any())).thenReturn(true);
+
+        KafkaFileDTO kafkaFileDTO = getKafkaFileDTO();
+        ResultStatus resultStatus = kafkaFileService.uploadKafkaFile(kafkaFileDTO, ADMIN);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.RESOURCE_ALREADY_EXISTED.getCode());
+    }
+
+
+    @Test
+    public void modifyKafkaFileTest() {
+        // paramIllegal
+        modifyKafkaFile2ParamIllegalTest();
+        // resource not exist
+        modifyKafkaFile2ResourceNotExistTest();
+        // operation failed
+        modifyKafkaFile2OperationFailedTest();
+        // mysqlError
+        modifyKafkaFile2MysqlErrorTest();
+        // success
+        modifyKafkaFile2SuccessTest();
+    }
+
+    private void modifyKafkaFile2ParamIllegalTest() {
+        ResultStatus resultStatus = kafkaFileService.modifyKafkaFile(null, ADMIN);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.PARAM_ILLEGAL.getCode());
+    }
+
+    private void modifyKafkaFile2ResourceNotExistTest() {
+        Mockito.when(kafkaFileDao.getById(Mockito.anyLong())).thenReturn(null);
+
+        KafkaFileDTO kafkaFileDTO = getKafkaFileDTO();
+        ResultStatus resultStatus = kafkaFileService.modifyKafkaFile(kafkaFileDTO, ADMIN);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.RESOURCE_NOT_EXIST.getCode());
+    }
+
+    private void modifyKafkaFile2OperationFailedTest() {
+        Mockito.when(kafkaFileDao.getById(Mockito.anyLong())).thenReturn(new KafkaFileDO());
+        Mockito.when(kafkaFileDao.updateById(Mockito.any())).thenReturn(-1);
+
+        KafkaFileDTO kafkaFileDTO = getKafkaFileDTO();
+        kafkaFileDTO.setFileType(-1);
+        ResultStatus resultStatus1 = kafkaFileService.modifyKafkaFile(kafkaFileDTO, ADMIN);
+        Assert.assertEquals(resultStatus1.getCode(), ResultStatus.OPERATION_FAILED.getCode());
+
+        kafkaFileDTO.setFileType(KafkaFileEnum.PACKAGE.getCode());
+        kafkaFileDTO.setFileName("xxx");
+        ResultStatus resultStatus2 = kafkaFileService.modifyKafkaFile(kafkaFileDTO, ADMIN);
+        Assert.assertEquals(resultStatus2.getCode(), ResultStatus.OPERATION_FAILED.getCode());
+    }
+
+    private void modifyKafkaFile2MysqlErrorTest() {
+        KafkaFileDO kafkaFileDO = getKafkaFileDO();
+        Mockito.when(kafkaFileDao.getById(Mockito.anyLong())).thenReturn(kafkaFileDO);
+        Mockito.when(kafkaFileDao.updateById(Mockito.any())).thenReturn(-1);
+
+        KafkaFileDTO kafkaFileDTO = getKafkaFileDTO();
+        ResultStatus resultStatus1 = kafkaFileService.modifyKafkaFile(kafkaFileDTO, ADMIN);
+        Assert.assertEquals(resultStatus1.getCode(), ResultStatus.MYSQL_ERROR.getCode());
+
+    }
+
+    private void modifyKafkaFile2SuccessTest() {
+        KafkaFileDO kafkaFileDO = getKafkaFileDO();
+        Mockito.when(kafkaFileDao.getById(Mockito.anyLong())).thenReturn(kafkaFileDO);
+        Mockito.when(kafkaFileDao.updateById(Mockito.any())).thenReturn(1);
+        Mockito.when(storageService.upload(Mockito.any(), Mockito.any(), Mockito.any())).thenReturn(true);
+
+        KafkaFileDTO kafkaFileDTO = getKafkaFileDTO();
+        ResultStatus resultStatus1 = kafkaFileService.modifyKafkaFile(kafkaFileDTO, ADMIN);
+        Assert.assertEquals(resultStatus1.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+    @Test
+    public void downloadKafkaFileTest() {
+        // resource not exist
+        downloadKafkaFile2ResourceNotExist();
+        // STORAGE_FILE_TYPE_NOT_SUPPORT
+        downloadKafkaFile2FileNotSupportExist();
+        // success
+        downloadKafkaFile2SuccessExist();
+    }
+
+    private void downloadKafkaFile2ResourceNotExist() {
+        Mockito.when(kafkaFileDao.getById(Mockito.anyLong())).thenReturn(null);
+        Result<MultipartFile> resultStatus = kafkaFileService.downloadKafkaFile(KAFKA_FILE_ID);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.RESOURCE_NOT_EXIST.getCode());
+    }
+
+    private void downloadKafkaFile2FileNotSupportExist() {
+        KafkaFileDO kafkaFileDO = getKafkaFileDO();
+        Mockito.when(kafkaFileDao.getById(Mockito.anyLong())).thenReturn(kafkaFileDO);
+        Result<MultipartFile> resultStatus = kafkaFileService.downloadKafkaFile(KAFKA_FILE_ID);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.STORAGE_FILE_TYPE_NOT_SUPPORT.getCode());
+    }
+
+    private void downloadKafkaFile2SuccessExist() {
+        Mockito.when(storageService.download(Mockito.any(), Mockito.any())).thenReturn(Result.buildFrom(ResultStatus.SUCCESS));
+        KafkaFileDO kafkaFileDO = getKafkaFileDO();
+        kafkaFileDO.setFileType(KafkaFileEnum.SERVER_CONFIG.getCode());
+        Mockito.when(kafkaFileDao.getById(Mockito.anyLong())).thenReturn(kafkaFileDO);
+
+        Result<MultipartFile> resultStatus = kafkaFileService.downloadKafkaFile(KAFKA_FILE_ID);
+        Assert.assertEquals(resultStatus.getCode(), ResultStatus.SUCCESS.getCode());
+    }
+
+
+}