diff --git a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/BrokerMetricESSender.java b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/BrokerMetricESSender.java
similarity index 79%
rename from km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/BrokerMetricESSender.java
rename to km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/BrokerMetricESSender.java
index 57d43ad8..1c074df5 100644
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/BrokerMetricESSender.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/BrokerMetricESSender.java
@@ -1,7 +1,8 @@
-package com.xiaojukeji.know.streaming.km.collector.sink;
+package com.xiaojukeji.know.streaming.km.collector.sink.kafka;
 
 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.collector.sink.AbstractMetricESSender;
 import com.xiaojukeji.know.streaming.km.common.bean.event.metric.BrokerMetricEvent;
 import com.xiaojukeji.know.streaming.km.common.bean.po.metrice.BrokerMetricPO;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
@@ -10,7 +11,7 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
 
-import static com.xiaojukeji.know.streaming.km.common.constant.ESIndexConstant.BROKER_INDEX;
+import static com.xiaojukeji.know.streaming.km.persistence.es.template.TemplateConstant.BROKER_INDEX;
 
 @Component
 public class BrokerMetricESSender extends AbstractMetricESSender implements ApplicationListener<BrokerMetricEvent> {
diff --git a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/ClusterMetricESSender.java b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/ClusterMetricESSender.java
similarity index 79%
rename from km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/ClusterMetricESSender.java
rename to km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/ClusterMetricESSender.java
index 478a27bd..70ee3d89 100644
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/ClusterMetricESSender.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/ClusterMetricESSender.java
@@ -1,7 +1,8 @@
-package com.xiaojukeji.know.streaming.km.collector.sink;
+package com.xiaojukeji.know.streaming.km.collector.sink.kafka;
 
 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.collector.sink.AbstractMetricESSender;
 import com.xiaojukeji.know.streaming.km.common.bean.event.metric.ClusterMetricEvent;
 import com.xiaojukeji.know.streaming.km.common.bean.po.metrice.ClusterMetricPO;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
@@ -10,8 +11,7 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
 
-
-import static com.xiaojukeji.know.streaming.km.common.constant.ESIndexConstant.CLUSTER_INDEX;
+import static com.xiaojukeji.know.streaming.km.persistence.es.template.TemplateConstant.CLUSTER_INDEX;
 
 @Component
 public class ClusterMetricESSender extends AbstractMetricESSender implements ApplicationListener<ClusterMetricEvent> {
diff --git a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/GroupMetricESSender.java b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/GroupMetricESSender.java
similarity index 79%
rename from km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/GroupMetricESSender.java
rename to km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/GroupMetricESSender.java
index e7e622c5..2192ad90 100644
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/GroupMetricESSender.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/GroupMetricESSender.java
@@ -1,7 +1,8 @@
-package com.xiaojukeji.know.streaming.km.collector.sink;
+package com.xiaojukeji.know.streaming.km.collector.sink.kafka;
 
 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.collector.sink.AbstractMetricESSender;
 import com.xiaojukeji.know.streaming.km.common.bean.event.metric.GroupMetricEvent;
 import com.xiaojukeji.know.streaming.km.common.bean.po.metrice.GroupMetricPO;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
@@ -10,8 +11,7 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
 
-
-import static com.xiaojukeji.know.streaming.km.common.constant.ESIndexConstant.GROUP_INDEX;
+import static com.xiaojukeji.know.streaming.km.persistence.es.template.TemplateConstant.GROUP_INDEX;
 
 @Component
 public class GroupMetricESSender extends AbstractMetricESSender implements ApplicationListener<GroupMetricEvent> {
diff --git a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/PartitionMetricESSender.java b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/PartitionMetricESSender.java
similarity index 80%
rename from km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/PartitionMetricESSender.java
rename to km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/PartitionMetricESSender.java
index 460d5e92..40087e28 100644
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/PartitionMetricESSender.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/PartitionMetricESSender.java
@@ -1,7 +1,8 @@
-package com.xiaojukeji.know.streaming.km.collector.sink;
+package com.xiaojukeji.know.streaming.km.collector.sink.kafka;
 
 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.collector.sink.AbstractMetricESSender;
 import com.xiaojukeji.know.streaming.km.common.bean.event.metric.PartitionMetricEvent;
 import com.xiaojukeji.know.streaming.km.common.bean.po.metrice.PartitionMetricPO;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
@@ -10,7 +11,7 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
 
-import static com.xiaojukeji.know.streaming.km.common.constant.ESIndexConstant.PARTITION_INDEX;
+import static com.xiaojukeji.know.streaming.km.persistence.es.template.TemplateConstant.PARTITION_INDEX;
 
 @Component
 public class PartitionMetricESSender extends AbstractMetricESSender implements ApplicationListener<PartitionMetricEvent> {
diff --git a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/ReplicaMetricESSender.java b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/ReplicaMetricESSender.java
similarity index 79%
rename from km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/ReplicaMetricESSender.java
rename to km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/ReplicaMetricESSender.java
index 9b39f3af..d7b74905 100644
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/ReplicaMetricESSender.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/ReplicaMetricESSender.java
@@ -1,7 +1,8 @@
-package com.xiaojukeji.know.streaming.km.collector.sink;
+package com.xiaojukeji.know.streaming.km.collector.sink.kafka;
 
 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.collector.sink.AbstractMetricESSender;
 import com.xiaojukeji.know.streaming.km.common.bean.event.metric.ReplicaMetricEvent;
 import com.xiaojukeji.know.streaming.km.common.bean.po.metrice.ReplicationMetricPO;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
@@ -10,7 +11,7 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
 
-import static com.xiaojukeji.know.streaming.km.common.constant.ESIndexConstant.REPLICATION_INDEX;
+import static com.xiaojukeji.know.streaming.km.persistence.es.template.TemplateConstant.REPLICATION_INDEX;
 
 @Component
 public class ReplicaMetricESSender extends AbstractMetricESSender implements ApplicationListener<ReplicaMetricEvent> {
diff --git a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/TopicMetricESSender.java b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/TopicMetricESSender.java
similarity index 79%
rename from km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/TopicMetricESSender.java
rename to km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/TopicMetricESSender.java
index 311a26fa..fb3cff7d 100644
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/TopicMetricESSender.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/TopicMetricESSender.java
@@ -1,7 +1,8 @@
-package com.xiaojukeji.know.streaming.km.collector.sink;
+package com.xiaojukeji.know.streaming.km.collector.sink.kafka;
 
 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.collector.sink.AbstractMetricESSender;
 import com.xiaojukeji.know.streaming.km.common.bean.event.metric.*;
 import com.xiaojukeji.know.streaming.km.common.bean.po.metrice.*;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
@@ -10,8 +11,7 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
 
-
-import static com.xiaojukeji.know.streaming.km.common.constant.ESIndexConstant.TOPIC_INDEX;
+import static com.xiaojukeji.know.streaming.km.persistence.es.template.TemplateConstant.TOPIC_INDEX;
 
 @Component
 public class TopicMetricESSender extends AbstractMetricESSender implements ApplicationListener<TopicMetricEvent> {
diff --git a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/ZookeeperMetricESSender.java b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/ZookeeperMetricESSender.java
similarity index 80%
rename from km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/ZookeeperMetricESSender.java
rename to km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/ZookeeperMetricESSender.java
index f2f254d6..c93f4860 100644
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/ZookeeperMetricESSender.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/kafka/ZookeeperMetricESSender.java
@@ -1,7 +1,8 @@
-package com.xiaojukeji.know.streaming.km.collector.sink;
+package com.xiaojukeji.know.streaming.km.collector.sink.kafka;
 
 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.collector.sink.AbstractMetricESSender;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
 import com.xiaojukeji.know.streaming.km.common.bean.event.metric.ZookeeperMetricEvent;
 import com.xiaojukeji.know.streaming.km.common.bean.po.metrice.ZookeeperMetricPO;
@@ -10,7 +11,7 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
 
-import static com.xiaojukeji.know.streaming.km.common.constant.ESIndexConstant.ZOOKEEPER_INDEX;
+import static com.xiaojukeji.know.streaming.km.persistence.es.template.TemplateConstant.ZOOKEEPER_INDEX;
 
 @Component
 public class ZookeeperMetricESSender extends AbstractMetricESSender implements ApplicationListener<ZookeeperMetricEvent> {