[#noissue] Change heatmap kafka key for distribution

Author: donghun-cho

collector/src/main/java/com/navercorp/pinpoint/collector/heatmap/config/HeatmapProperties.java

@@ -31,7 +31,8 @@ public class HeatmapProperties {
     private int heatMapTopicPaddingLength;
     @Value("${kafka.heatmap.topic.count}")
     private int heatmapTopicCount;
-
+    @Value("${kafka.heatmap.key.partition.count}")
+    private int heatmapKeyPartitionCount;
 
     public String getHeatmapTopicPrefix() {
         return heatmapTopicPrefix;
@@ -44,4 +45,8 @@ public int getHeatMapTopicPaddingLength() {
     public int getHeatmapTopicCount() {
         return heatmapTopicCount;
     }
+
+    public int getHeatmapKeyPartitionCount() {
+        return heatmapKeyPartitionCount;
+    }
 }

collector/src/main/java/com/navercorp/pinpoint/collector/heatmap/dao/pinot/PinotHeatmapDao.java

@@ -18,14 +18,17 @@
 
 import com.navercorp.pinpoint.collector.heatmap.config.HeatmapProperties;
 import com.navercorp.pinpoint.collector.heatmap.dao.HeatmapDao;
+import com.navercorp.pinpoint.collector.heatmap.util.HashmapSortKeyUtils;
 import com.navercorp.pinpoint.collector.heatmap.vo.HeatmapStat;
 import com.navercorp.pinpoint.common.server.metric.dao.TopicNameManager;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Repository;
 
 import java.util.Objects;
+import java.util.concurrent.ThreadLocalRandom;
 
 /**
  * @author minwoo-jung
@@ -36,10 +39,12 @@ public class PinotHeatmapDao implements HeatmapDao {
     private final Logger logger = LogManager.getLogger(getClass());
     private final KafkaTemplate<String, HeatmapStat> kafkaHeatmapStatTemplate;
     private final TopicNameManager topicNameManager;
+    private final int keyPartitionCount;
 
     public PinotHeatmapDao(KafkaTemplate<String, HeatmapStat> kafkaHeatmapStatTemplate, HeatmapProperties heatmapProperties) {
         this.kafkaHeatmapStatTemplate = Objects.requireNonNull(kafkaHeatmapStatTemplate, "kafkaHeatmapStatTemplate");
         this.topicNameManager = new TopicNameManager(heatmapProperties.getHeatmapTopicPrefix(), heatmapProperties.getHeatMapTopicPaddingLength(), heatmapProperties.getHeatmapTopicCount());
+        this.keyPartitionCount = heatmapProperties.getHeatmapKeyPartitionCount();
     }
 
     @Override
@@ -49,6 +54,14 @@ public void insert(HeatmapStat heatmapStat) {
             return;
         }
         String topic = topicNameManager.getTopicName(heatmapStat.getApplicationName());
-        kafkaHeatmapStatTemplate.send(topic, heatmapStat.getAgentId(), heatmapStat);
+        kafkaHeatmapStatTemplate.send(topic, heatmapStat.getSortKey() + "#" + randomPartitionSuffix(), heatmapStat);
+    }
+
+    private int randomPartitionSuffix() {
+        if (keyPartitionCount <= 1) {
+            return 0;
+        }
+        int nonNegativeRandom = ThreadLocalRandom.current().nextInt() & 0x7fffffff;
+        return nonNegativeRandom % keyPartitionCount;
     }
 }

collector/src/main/resources/heatmap/collector/profiles/local/heatmap-collector.properties

@@ -1,3 +1,4 @@
 kafka.heatmap.topic.count=1
 kafka.heatmap.topic.prefix=heatmap-stat-app-
-kafka.heatmap.topic.padding.length=2
+kafka.heatmap.topic.padding.length=2
+kafka.heatmap.key.partition.count=1

collector/src/main/resources/heatmap/collector/profiles/release/heatmap-collector.properties

@@ -1,3 +1,4 @@
 kafka.heatmap.topic.count=1
 kafka.heatmap.topic.prefix=heatmap-stat-app-
-kafka.heatmap.topic.padding.length=2
+kafka.heatmap.topic.padding.length=2
+kafka.heatmap.key.partition.count=1