Add config topic-management.minPartitionNum (linkedin#73)

Author: lindong28

README.md

@@ -48,7 +48,7 @@ different Kafka clusters that are connected by MirrorMaker to monitor their
 end-to-end latency.
 
 - Kafka Monitor by default will automatically create the monitor topic based on
-the e.g.  `topic-management.replicationFactor` and `topic-management.partitionsToBrokerRatio`
+the e.g.  `topic-management.replicationFactor` and `topic-management.partitionsToBrokersRatio`
 specified in the config. replicationFactor is 1 by default and you probably
 want to change it to the same replication factor as used for your existing
 topics. You can disable auto topic creation by setting `produce.topic.topicCreationEnabled` to false.

src/main/java/com/linkedin/kmf/services/MultiClusterTopicManagementService.java

@@ -53,7 +53,15 @@
  * This service periodically checks and rebalances the monitor topics across a pipeline of Kafka clusters so that
  * leadership of the partitions of the monitor topic in each cluster is distributed evenly across brokers in the cluster.
  *
- * It also makes sure that the number of partitions of the monitor topics is same across the monitored clusters.
+ * More specifically, this service may do some or all of the following tasks depending on the config:
+ *
+ * - Create the monitor topic using the user-specified replication factor and partition number
+ * - Increase partition number of the monitor topic if either partitionsToBrokersRatio or minPartitionNum is not satisfied
+ * - Increase replication factor of the monitor topic if the user-specified replicationFactor is not satisfied
+ * - Reassign partition across brokers to make sure each broker acts as preferred leader of at least one partition of the monitor topic
+ * - Trigger preferred leader election to make sure each broker acts as leader of at least one partition of the monitor topic.
+ * - Make sure the number of partitions of the monitor topic is same across all monitored custers.
+ *
  */
 public class MultiClusterTopicManagementService implements Service {
   private static final Logger LOG = LoggerFactory.getLogger(MultiClusterTopicManagementService.class);
@@ -134,6 +142,13 @@ public void run() {
           helper.maybeCreateTopic();
         }
 
+        /*
+         * The partition number of the monitor topics should be the minimum partition number that satisifies the following conditions:
+         * - partition number of the monitor topics across all monitored clusters should be the same
+         * - partitionNum / brokerNum >= user-configured partitionsToBrokersRatio.
+         * - partitionNum >= user-configured minPartitionNum
+         */
+
         int minPartitionNum = 0;
         for (TopicManagementHelper helper : _topicManagementByCluster.values()) {
           minPartitionNum = Math.max(minPartitionNum, helper.minPartitionNum());
@@ -164,7 +179,8 @@ static class TopicManagementHelper {
     private final String _topic;
     private final String _zkConnect;
     private final int _replicationFactor;
-    private final double _partitionsToBrokerRatio;
+    private final double _minPartitionsToBrokersRatio;
+    private final int _minPartitionNum;
     private final TopicFactory _topicFactory;
 
     TopicManagementHelper(Map<String, Object> props) throws Exception {
@@ -173,7 +189,8 @@ static class TopicManagementHelper {
       _topic = config.getString(TopicManagementServiceConfig.TOPIC_CONFIG);
       _zkConnect = config.getString(TopicManagementServiceConfig.ZOOKEEPER_CONNECT_CONFIG);
       _replicationFactor = config.getInt(TopicManagementServiceConfig.TOPIC_REPLICATION_FACTOR_CONFIG);
-      _partitionsToBrokerRatio = config.getDouble(TopicManagementServiceConfig.PARTITIONS_TO_BROKERS_RATIO_CONFIG);
+      _minPartitionsToBrokersRatio = config.getDouble(TopicManagementServiceConfig.PARTITIONS_TO_BROKERS_RATIO_CONFIG);
+      _minPartitionNum = config.getInt(TopicManagementServiceConfig.MIN_PARTITION_NUM_CONFIG);
       String topicFactoryClassName = config.getString(TopicManagementServiceConfig.TOPIC_FACTORY_CLASS_CONFIG);
       Map topicFactoryConfig = props.containsKey(TopicManagementServiceConfig.TOPIC_FACTORY_PROPS_CONFIG) ?
           (Map) props.get(TopicManagementServiceConfig.TOPIC_FACTORY_PROPS_CONFIG) : new HashMap();
@@ -182,13 +199,13 @@ static class TopicManagementHelper {
 
     void maybeCreateTopic() throws Exception {
       if (_topicCreationEnabled) {
-        _topicFactory.createTopicIfNotExist(_zkConnect, _topic, _replicationFactor, _partitionsToBrokerRatio, new Properties());
+        _topicFactory.createTopicIfNotExist(_zkConnect, _topic, _replicationFactor, _minPartitionsToBrokersRatio, new Properties());
       }
     }
 
     int minPartitionNum() {
       int brokerCount = Utils.getBrokerCount(_zkConnect);
-      return (int) Math.ceil(_partitionsToBrokerRatio * brokerCount);
+      return Math.max((int) Math.ceil(_minPartitionsToBrokersRatio * brokerCount), _minPartitionNum);
     }
 
     void maybeAddPartitions(int minPartitionNum) {

src/main/java/com/linkedin/kmf/services/configs/TopicManagementServiceConfig.java

@@ -32,18 +32,21 @@ public class TopicManagementServiceConfig extends AbstractConfig {
   public static final String TOPIC_DOC = CommonServiceConfig.TOPIC_DOC;
 
   public static final String PARTITIONS_TO_BROKERS_RATIO_CONFIG = "topic-management.partitionsToBrokersRatio";
-  public static final String PARTITIONS_TO_BROKERS_RATIO_DOC = "New partitions are created when the actual ratio falls below this threshold."
-      + " partitionsToBrokerRatio will be the lower bound on the number of partitions per broker when a topic is"
-      + " created or when the partition is added.";
+  public static final String PARTITIONS_TO_BROKERS_RATIO_DOC = "New partitions are added when the actual ratio falls below this threshold."
+      + " This config provides a loose lower bound on the partitionNum-to-brokerNum ratio when the monitor topic is created or when partition is added.";
+
+  public static final String MIN_PARTITION_NUM_CONFIG = "topic-management.minPartitionNum";
+  public static final String MIN_PARTITION_NUM_DOC = "New partitions are added when the actual partition number falls below this threshold."
+      + " This config provides a loose lower bound on the partition number of the monitor topic when the topic is created or when partition is added.";
 
   public static final String TOPIC_REPLICATION_FACTOR_CONFIG = "topic-management.replicationFactor";
   public static final String TOPIC_REPLICATION_FACTOR_DOC = "When a topic is created automatically this is the "
       + "replication factor used.";
 
   public static final String TOPIC_CREATION_ENABLED_CONFIG = "topic-management.topicCreationEnabled";
-  public static final String TOPIC_CREATION_ENABLED_DOC = "When true this automatically creates the topic mentioned by \""
-    + CommonServiceConfig.TOPIC_CONFIG + "\" with replication factor \"" + TOPIC_REPLICATION_FACTOR_CONFIG + "and min ISR of max("
-    + TOPIC_REPLICATION_FACTOR_CONFIG + "-1, 1) with number of brokers * \"" + PARTITIONS_TO_BROKERS_RATIO_CONFIG + "\" partitions.";
+  public static final String TOPIC_CREATION_ENABLED_DOC = String.format("When true this service automatically creates the topic named"
+      + " in the config with replication factor %s and min ISR as max(%s - 1, 1). The partition number is determined based on %s and %s",
+      TOPIC_REPLICATION_FACTOR_CONFIG, TOPIC_REPLICATION_FACTOR_CONFIG, PARTITIONS_TO_BROKERS_RATIO_CONFIG, MIN_PARTITION_NUM_DOC);
 
   public static final String TOPIC_FACTORY_CLASS_CONFIG = "topic-management.topicFactory.class.name";
   public static final String TOPIC_FACTORY_CLASS_DOC = "The name of the class used to create topics.  This class must implement "
@@ -64,8 +67,14 @@ public class TopicManagementServiceConfig extends AbstractConfig {
               TOPIC_DOC)
       .define(PARTITIONS_TO_BROKERS_RATIO_CONFIG,
               ConfigDef.Type.DOUBLE,
-              1.5,
-              ConfigDef.Importance.LOW, PARTITIONS_TO_BROKERS_RATIO_DOC)
+              1.0,
+              ConfigDef.Importance.LOW,
+              PARTITIONS_TO_BROKERS_RATIO_DOC)
+      .define(MIN_PARTITION_NUM_CONFIG,
+              ConfigDef.Type.INT,
+              1,
+              ConfigDef.Importance.LOW,
+              MIN_PARTITION_NUM_DOC)
       .define(TOPIC_CREATION_ENABLED_CONFIG,
               ConfigDef.Type.BOOLEAN,
               true,