采样调整

Author: zengqiao

km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/topic/impl/TopicStateManagerImpl.java

@@ -142,12 +142,13 @@ public Result<List<TopicRecordVO>> getTopicMessages(Long clusterPhyId, String to
             // 创建kafka-consumer
             kafkaConsumer = new KafkaConsumer<>(this.generateClientProperties(clusterPhy, dto.getMaxRecords()));
 
+            kafkaConsumer.assign(endOffsetsMapResult.getData().keySet());
+            for (Map.Entry<TopicPartition, Long> entry: endOffsetsMapResult.getData().entrySet()) {
+                kafkaConsumer.seek(entry.getKey(), Math.max(0, entry.getValue() - dto.getMaxRecords()));
+            }
+
             // 这里需要减去 KafkaConstant.POLL_ONCE_TIMEOUT_UNIT_MS 是因为poll一次需要耗时，如果这里不减去，则可能会导致poll之后，超过要求的时间
             while (System.currentTimeMillis() - startTime + KafkaConstant.POLL_ONCE_TIMEOUT_UNIT_MS <= dto.getPullTimeoutUnitMs() && voList.size() < dto.getMaxRecords()) {
-                for (Map.Entry<TopicPartition, Long> entry: endOffsetsMapResult.getData().entrySet()) {
-                    kafkaConsumer.assign(Arrays.asList(entry.getKey()));
-                    kafkaConsumer.seek(entry.getKey(), Math.max(0, entry.getValue() - dto.getMaxRecords()));
-
                     ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(KafkaConstant.POLL_ONCE_TIMEOUT_UNIT_MS));
                     for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                         if (this.checkIfIgnore(consumerRecord, dto.getFilterKey(), dto.getFilterValue())) {
@@ -165,7 +166,6 @@ public Result<List<TopicRecordVO>> getTopicMessages(Long clusterPhyId, String to
                             || voList.size() > dto.getMaxRecords()) {
                         break;
                     }
-                }
             }
 
             return Result.buildSuc(voList.subList(0, Math.min(dto.getMaxRecords(), voList.size())));

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/entity/record/RecordHeaderKS.java

@@ -10,5 +10,5 @@
 public class RecordHeaderKS {
     private String key;
 
-    private byte[] value;
+    private String value;
 }

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/converter/TopicVOConverter.java

@@ -18,6 +18,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.header.Header;
 
+import java.nio.charset.StandardCharsets;
 import java.util.*;
 
 public class TopicVOConverter {
@@ -51,7 +52,7 @@ public static TopicRecordVO convert2TopicRecordVO(String topicName, ConsumerReco
         vo.setValue(consumerRecord.value());
         vo.setHeaderList(new ArrayList<>());
         for (Header header : consumerRecord.headers().toArray()) {
-            vo.getHeaderList().add(new RecordHeaderKS(header.key(), header.value()));
+            vo.getHeaderList().add(new RecordHeaderKS(header.key(), new String(header.value(), StandardCharsets.UTF_8)));
         }
         return vo;
     }