Merge pull request #19 from navdeepsekhon/18-preserve-partitions

#18 option to preserve partition count

Author: navdeepsekhon

README.md

@@ -14,6 +14,7 @@
  - [Delete created topics (--wipe)](#delete-created-topics)
  - [Delete schemas from schema registry (--wipe-schemas)](#delete-schemas)
  - [Debug (--debug)](#debug)
+ - [Preserve Partition Count (--preserve-partition-count)](#preserve-partition-count)
  - [Contributions](#contributions)
 
 
@@ -122,6 +123,8 @@ If the partitions listed in config are less than the existing - an exception wil
 
 If they are same - nothing.
 
+If flag `--preserve-partition-count` is used, partitions will not be updated.
+
 ### All other configs:
 All other configs will be updated to the new values from config.
 
@@ -216,6 +219,10 @@ kafkaer.schema.registry.ssl.truststore.location=...
 
 Use flag `--debug` for detailed logging
 
+# Preserve Partition Count
+
+If a topic already exists and it's partition count is different from what is defined in the config, kafkaer will try update the partitions as described above. In order to ignore the partition count and keep the existing partitions, `--preserve-partition-count` flag can be used. When used, the difference is partition count will only be logged.
+
 # Contributions
 Merge requests welcome. Please create an issue with change details and link it to your merge request.
 

build.gradle

@@ -6,7 +6,7 @@ apply plugin: 'maven'
 apply plugin: 'signing'
 
 group 'co.navdeep'
-version '1.4.1'
+version '1.4.2'
 archivesBaseName = "kafkaer"
 
 sourceCompatibility = 1.8

src/main/java/co/navdeep/kafkaer/App.java

@@ -28,8 +28,9 @@ public static void main(String[] a) throws Exception {
             throw new RuntimeException("Missing required arguments - propertiesLocation, configLocation");
         }
 
-        logger.debug("Input args: config: [{}] properties: [{}] wipe:[{}] confirm-delete: [{}], wipe-schema: [{}]", args.getConfig(), args.getProperties(), args.isWipe(), args.isConfirmDelete(), args.isWipeSchemas());
+        logger.debug("Input args: config: [{}] properties: [{}] wipe:[{}] confirm-delete: [{}], wipe-schema: [{}], preserve-partition-count: [{}]", args.getConfig(), args.getProperties(), args.isWipe(), args.isConfirmDelete(), args.isWipeSchemas(), args.isPreservePartitionCount());
         Configurator configurator = new Configurator(args.getProperties(), args.getConfig());
+        configurator.setPreservePartitionCount(args.isPreservePartitionCount());
         if(args.isWipe())
             configurator.wipeTopics(args.isConfirmDelete(), args.isWipeSchemas());
         else

src/main/java/co/navdeep/kafkaer/Args.java

@@ -27,6 +27,9 @@ public class Args {
     @Option(name="--confirm-delete", usage="Used with --wipe. Will wait for all brokers to sync up to ensure topic is deleted from all. Default max wait 60s. Configure using " + Utils.MAX_DELETE_CONFIRM_WAIT_CONFIG, handler = BooleanOptionHandler.class)
     boolean confirmDelete;
 
+    @Option(name="--preserve-partition-count", usage="If a topic already exists and it's partition count is different from config, the partition count will not be changed.", handler = BooleanOptionHandler.class)
+    boolean preservePartitionCount;
+
     @Option(name="--help", aliases= "-h", help = true, usage="list usage", handler =  BooleanOptionHandler.class)
     boolean help;
 

src/main/java/co/navdeep/kafkaer/Configurator.java

@@ -29,6 +29,7 @@ public class Configurator {
     private Config config;
     private AdminClient adminClient;
     private SchemaRegistryClient schemaRegistryClient;
+    private boolean preservePartitionCount;
 
     private static Logger logger = LoggerFactory.getLogger(Configurator.class);
 
@@ -194,7 +195,21 @@ private void handleTopicConfigUpdate(Topic topic) throws InterruptedException {
             throw new RuntimeException(e);
         }
     }
+
     private void handleTopicPartitionsUpdate(TopicDescription current, Topic topic) throws InterruptedException {
+        if(preservePartitionCount) logPartitionDiff(current, topic);
+        else updatePartitions(current, topic);
+    }
+
+    private void logPartitionDiff(TopicDescription current, Topic topic){
+        if(current.partitions().size() < topic.getPartitions()){
+            logger.warn("Current partition count for topic {} is [{}], partition count in config is [{}]. Execute without --preserve-partition-count to make this partition update.", topic.getName(), current.partitions().size(), topic.getPartitions());
+
+        } else if(current.partitions().size() > topic.getPartitions()){
+            logger.warn("Current partition count for topic {} is [{}], partition count in config is [{}].", topic.getName(), current.partitions().size(), topic.getPartitions());
+        }
+    }
+    private void updatePartitions(TopicDescription current, Topic topic){
         try {
             if(current.partitions().size() < topic.getPartitions()){
                 logger.debug("Updating partition count for topic {} from [{}] to [{}]", topic.getName(), current.partitions().size(), topic.getPartitions());
@@ -203,7 +218,7 @@ private void handleTopicPartitionsUpdate(TopicDescription current, Topic topic)
             } else if(current.partitions().size() > topic.getPartitions()){
                 throw new RuntimeException("Can not reduce number of partitions for topic [" + topic.getName() + "] from current:" + current.partitions().size() + " to " + topic.getPartitions());
             }
-        } catch(ExecutionException e){
+        } catch(ExecutionException | InterruptedException e){
             throw new RuntimeException(e);
         }
     }

src/test/java/ConfiguratorTest.java

@@ -140,6 +140,29 @@ public void testIncreasePartitions() throws ExecutionException, InterruptedExcep
         compareWithKafkaTopic(topic);
     }
 
+    @Test
+    public void testPreservePartitions() throws ExecutionException, InterruptedException, ConfigurationException {
+        Config config = new Config();
+        String topicName = UUID.randomUUID().toString();
+        Topic topic = new Topic(topicName, 1, (short)1);
+        config.getTopics().add(topic);
+
+        Configurator configurator = new Configurator(Utils.readProperties(PROPERTIES_LOCATION), config);
+        configurator.setPreservePartitionCount(true);
+        configurator.applyConfig();
+
+        sleep();
+        compareWithKafkaTopic(topic);
+
+        //Increase the partitions and apply config
+        topic.setPartitions(2);
+        configurator.applyConfig();
+
+        //Still expect 1 partition
+        topic.setPartitions(1);
+        compareWithKafkaTopic(topic);
+    }
+
     @Test
     public void testUpdateExistingTopicConfig() throws ConfigurationException, ExecutionException, InterruptedException {
         Config config = new Config();