日志错误信息中补充Topic名称信息

Author: zengqiao

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/entity/param/partition/PartitionOffsetParam.java

@@ -1,6 +1,6 @@
 package com.xiaojukeji.know.streaming.km.common.bean.entity.param.partition;
 
-import com.xiaojukeji.know.streaming.km.common.bean.entity.param.cluster.ClusterPhyParam;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic.TopicParam;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import org.apache.kafka.clients.admin.OffsetSpec;
@@ -10,13 +10,13 @@
 
 @Data
 @NoArgsConstructor
-public class PartitionOffsetParam extends ClusterPhyParam {
+public class PartitionOffsetParam extends TopicParam {
     private Map<TopicPartition, OffsetSpec> topicPartitionOffsets;
 
     private Long timestamp;
 
-    public PartitionOffsetParam(Long clusterPhyId, Map<TopicPartition, OffsetSpec> topicPartitionOffsets, Long timestamp) {
-        super(clusterPhyId);
+    public PartitionOffsetParam(Long clusterPhyId, String topicName, Map<TopicPartition, OffsetSpec> topicPartitionOffsets, Long timestamp) {
+        super(clusterPhyId, topicName);
         this.topicPartitionOffsets = topicPartitionOffsets;
         this.timestamp = timestamp;
     }

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/entity/param/topic/TopicParam.java

@@ -15,4 +15,12 @@ public TopicParam(Long clusterPhyId, String topicName) {
         super(clusterPhyId);
         this.topicName = topicName;
     }
+
+    @Override
+    public String toString() {
+        return "TopicParam{" +
+                "clusterPhyId=" + clusterPhyId +
+                ", topicName='" + topicName + '\'' +
+                '}';
+    }
 }

km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/partition/impl/PartitionServiceImpl.java

@@ -207,7 +207,7 @@ public Result<Map<TopicPartition, Long>> getPartitionOffsetFromKafka(Long cluste
                 .forEach(elem -> topicPartitionOffsets.put(new TopicPartition(topicName, elem.getPartitionId()), offsetSpec));
 
         try {
-            return (Result<Map<TopicPartition, Long>>) doVCHandler(clusterPhyId, PARTITION_OFFSET_GET, new PartitionOffsetParam(clusterPhyId, topicPartitionOffsets, timestamp));
+            return (Result<Map<TopicPartition, Long>>) doVCHandler(clusterPhyId, PARTITION_OFFSET_GET, new PartitionOffsetParam(clusterPhyId, topicName, topicPartitionOffsets, timestamp));
         } catch (VCHandlerNotExistException e) {
             return Result.buildFailure(VC_HANDLE_NOT_EXIST);
         }
@@ -226,7 +226,7 @@ public Result<Map<TopicPartition, Long>> getPartitionOffsetFromKafka(Long cluste
                 .forEach(elem -> topicPartitionOffsets.put(new TopicPartition(topicName, elem.getPartitionId()), offsetSpec));
 
         try {
-            return (Result<Map<TopicPartition, Long>>) doVCHandler(clusterPhyId, PARTITION_OFFSET_GET, new PartitionOffsetParam(clusterPhyId, topicPartitionOffsets, timestamp));
+            return (Result<Map<TopicPartition, Long>>) doVCHandler(clusterPhyId, PARTITION_OFFSET_GET, new PartitionOffsetParam(clusterPhyId, topicName, topicPartitionOffsets, timestamp));
         } catch (VCHandlerNotExistException e) {
             return Result.buildFailure(VC_HANDLE_NOT_EXIST);
         }
@@ -300,7 +300,10 @@ private Result<Map<TopicPartition, Long>> getPartitionOffsetFromKafkaAdminClient
         } catch (NotExistException nee) {
             return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getClusterPhyNotExist(offsetParam.getClusterPhyId()));
         } catch (Exception e) {
-            log.error("method=getPartitionOffsetFromKafkaAdminClient||clusterPhyId={}||errMsg=exception!", offsetParam.getClusterPhyId(), e);
+            log.error(
+                    "class=PartitionServiceImpl||method=getPartitionOffsetFromKafkaAdminClient||clusterPhyId={}||topicName={}||errMsg=exception!",
+                    offsetParam.getClusterPhyId(), offsetParam.getTopicName(), e
+            );
 
             return Result.buildFromRSAndMsg(ResultStatus.KAFKA_OPERATE_FAILED, e.getMessage());
         }
@@ -355,7 +358,10 @@ private Result<Map<TopicPartition, Long>> getPartitionOffsetFromKafkaConsumerCli
         } catch (NotExistException nee) {
             return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getClusterPhyNotExist(offsetParam.getClusterPhyId()));
         } catch (Exception e) {
-            log.error("method=getPartitionOffsetFromKafkaConsumerClient||clusterPhyId={}||errMsg=exception!", offsetParam.getClusterPhyId(), e);
+            log.error(
+                    "class=PartitionServiceImpl||method=getPartitionOffsetFromKafkaConsumerClient||clusterPhyId={}||topicName={}||errMsg=exception!",
+                    offsetParam.getClusterPhyId(), offsetParam.getTopicName(), e
+            );
 
             return Result.buildFromRSAndMsg(ResultStatus.KAFKA_OPERATE_FAILED, e.getMessage());
         } finally {