Fix PartitionSizeAnomalyFinder, to be able to handle custom SELF_HEALING_PARTITION_SIZE_THRESHOLD_MB values

Author: k0b3rIT

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/PartitionSizeAnomalyFinder.java

@@ -108,9 +108,12 @@ 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);
-    _partitionSizeThresholdInMb = partitionSizeThreshold == null ? DEFAULT_SELF_HEALING_PARTITION_SIZE_THRESHOLD_MB
-                                                                 : partitionSizeThreshold;
+    String partitionSizeThresholdStr = (String) configs.get(SELF_HEALING_PARTITION_SIZE_THRESHOLD_MB_CONFIG);
+    _partitionSizeThresholdInMb = DEFAULT_SELF_HEALING_PARTITION_SIZE_THRESHOLD_MB;
+    if (partitionSizeThresholdStr != null) {
+      _partitionSizeThresholdInMb = Integer.parseInt(partitionSizeThresholdStr.trim());
+    }
+
     String topicPartitionSizeAnomalyClass = (String) configs.get(TOPIC_PARTITION_SIZE_ANOMALY_CLASS_CONFIG);
     if (topicPartitionSizeAnomalyClass == null) {
       _topicPartitionSizeAnomalyClass = DEFAULT_TOPIC_PARTITION_SIZE_ANOMALY_CLASS;