Fix PartitionSizeAnomalyFinder, to be able to handle custom SELF_HEALING_PARTITION_SIZE_THRESHOLD_MB values (#2212)

Author: k0b3rIT

cruise-control/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;