Fix PartitionSizeAnomalyFinder, to be able to handle custom SELF_HEAL…

…ING_PARTITION_SIZE_THRESHOLD_MB values (#2212)
linkedin · Jan 10, 2025 · 1af2863 · 1af2863
1 parent 6b5fe9e
commit 1af2863
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/...l/src/main/java/com/linkedin/kafka/cruisecontrol/detector/PartitionSizeAnomalyFinder.java b/...l/src/main/java/com/linkedin/kafka/cruisecontrol/detector/PartitionSizeAnomalyFinder.java
@@ -108,9 +108,9 @@ public void configure(Map<String, ?> configs) {
     String topicExcludedFromCheck = (String) configs.get(TOPIC_EXCLUDED_FROM_PARTITION_SIZE_CHECK);
     _topicExcludedFromCheck = Pattern.compile(topicExcludedFromCheck == null ? DEFAULT_TOPIC_EXCLUDED_FROM_PARTITION_SIZE_CHECK
                                                                              : topicExcludedFromCheck);
-    Integer partitionSizeThreshold = (Integer) configs.get(SELF_HEALING_PARTITION_SIZE_THRESHOLD_MB_CONFIG);
+    String partitionSizeThreshold = (String) configs.get(SELF_HEALING_PARTITION_SIZE_THRESHOLD_MB_CONFIG);
     _partitionSizeThresholdInMb = partitionSizeThreshold == null ? DEFAULT_SELF_HEALING_PARTITION_SIZE_THRESHOLD_MB
-                                                                 : partitionSizeThreshold;
+                                                                 : Integer.parseInt(partitionSizeThreshold);
     String topicPartitionSizeAnomalyClass = (String) configs.get(TOPIC_PARTITION_SIZE_ANOMALY_CLASS_CONFIG);
     if (topicPartitionSizeAnomalyClass == null) {
       _topicPartitionSizeAnomalyClass = DEFAULT_TOPIC_PARTITION_SIZE_ANOMALY_CLASS;